fix: Attach error handler to RPC stream (#3898)

When the execution environment fails on mobile (we still don't know the
cause of this) we were seeing uncaught "Premature close" errors, which
ends up crashing the app.

The issue is that all streams are destroyed when the Snap terminates (as
it does if it cannot initialize). Most streams already have error
handling setup via `pipeline` and if the Snap in question has already
run this applies to the RPC stream too. The problem is that if the Snap
never starts, the RPC stream also never has any error handlers attached,
so instead it bubbles up the error.

This PR fixes that by attaching an error handler to the RPC stream after
it has been created.

https://consensyssoftware.atlassian.net/browse/WPC-516

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Touches snap execution stream lifecycle in `AbstractExecutionService`,
which is central to snap RPC plumbing; while the change is small, it
could affect error propagation/logging during initialization and
teardown.
> 
> **Overview**
> Prevents uncaught RPC stream errors during snap startup/termination by
attaching an `error` listener to the JSON-RPC mux stream as soon as it’s
created, filtering out expected `Premature close` noise while logging
other failures.
> 
> Adds a regression test that simulates a failed/unreachable execution
environment and asserts the request fails cleanly without throwing
`Premature close`, and relaxes test utilities to accept any
`AbstractExecutionService` instance for injection.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
c8458fa. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: FrederikBolding

packages/snaps-controllers/coverage.json

@@ -1,5 +1,5 @@
 {
-  "branches": 95.16,
+  "branches": 94.97,
   "functions": 98.78,
   "lines": 98.63,
   "statements": 98.43

packages/snaps-controllers/src/services/AbstractExecutionService.ts

@@ -320,6 +320,14 @@ export abstract class AbstractExecutionService<WorkerType>
       }
     });
 
+    // An error handler is not attached to the RPC stream until `setupSnapProvider` is called.
+    // We must set it up here to prevent errors from bubbling up if the stream is destroyed before then.
+    rpcStream.on('error', (error) => {
+      if (error && !error.message?.match('Premature close')) {
+        logError(`Snap: "${snapId}" - RPC stream failure:`, error);
+      }
+    });
+
     const originalWrite = rpcStream.write.bind(rpcStream);
 
     // @ts-expect-error Hack to inspect the messages being written to the stream.

packages/snaps-controllers/src/snaps/SnapController.test.tsx

@@ -13,6 +13,7 @@ import type {
   CaveatConstraint,
 } from '@metamask/permission-controller';
 import { SubjectType } from '@metamask/permission-controller';
+import type { BasePostMessageStream } from '@metamask/post-message-stream';
 import { providerErrors, rpcErrors } from '@metamask/rpc-errors';
 import {
   WALLET_SNAP_PERMISSION_KEY,
@@ -59,6 +60,7 @@ import {
   DEFAULT_ICON_PATH,
   TEST_SECRET_RECOVERY_PHRASE_SEED_BYTES,
   MOCK_INITIAL_PERMISSIONS,
+  MockWindowPostMessageStream,
 } from '@metamask/snaps-utils/test-utils';
 import type { SemVerRange, SemVerVersion, Json } from '@metamask/utils';
 import {
@@ -93,8 +95,12 @@ import {
   SNAP_APPROVAL_RESULT,
   SNAP_APPROVAL_UPDATE,
 } from './SnapController';
-import { setupMultiplex } from '../services';
-import type { NodeThreadExecutionService } from '../services/node';
+import { AbstractExecutionService, setupMultiplex } from '../services';
+import type {
+  ExecutionServiceMessenger,
+  NodeThreadExecutionService,
+  TerminateJobArgs,
+} from '../services/node';
 import type { SnapControllerStateWithStorageService } from '../test-utils';
 import {
   approvalControllerMock,
@@ -1854,6 +1860,56 @@ describe('SnapController', () => {
     snapController.destroy();
   });
 
+  // This test also ensures that we do not throw "Premature close"
+  it('throws if the execution environment fails', async () => {
+    const rootMessenger = getControllerMessenger();
+    const options = getSnapControllerOptions({
+      rootMessenger,
+      state: { snaps: getPersistedSnapsState() },
+    });
+
+    class BrickedExecutionService extends AbstractExecutionService<null> {
+      constructor(messenger: ExecutionServiceMessenger) {
+        super({ messenger, setupSnapProvider: jest.fn(), pingTimeout: 1 });
+      }
+
+      protected async terminateJob(
+        _job: TerminateJobArgs<null>,
+      ): Promise<void> {
+        // no-op
+      }
+
+      protected async initEnvStream(
+        _snapId: string,
+      ): Promise<{ worker: null; stream: BasePostMessageStream }> {
+        return { worker: null, stream: new MockWindowPostMessageStream() };
+      }
+    }
+
+    const [snapController] = await getSnapControllerWithEES(
+      options,
+      new BrickedExecutionService(getNodeEESMessenger(rootMessenger)),
+    );
+
+    await expect(
+      snapController.handleRequest({
+        snapId: MOCK_SNAP_ID,
+        origin: MOCK_ORIGIN,
+        handler: HandlerType.OnRpcRequest,
+        request: {
+          jsonrpc: '2.0',
+          method: 'test',
+          params: {},
+          id: 1,
+        },
+      }),
+    ).rejects.toThrow(
+      'The executor for "npm:@metamask/example-snap" was unreachable. The executor did not respond in time.',
+    );
+
+    snapController.destroy();
+  });
+
   it('throws if unresponsive Snap is terminated while executing', async () => {
     const { manifest, sourceCode, svgIcon } =
       await getMockSnapFilesWithUpdatedChecksum({

packages/snaps-controllers/src/test-utils/controller.tsx

@@ -68,6 +68,7 @@ import type {
   MultichainRouterAllowedActions,
   MultichainRouterEvents,
 } from '../multichain';
+import type { AbstractExecutionService } from '../services';
 import type {
   AllowedActions,
   AllowedEvents,
@@ -691,7 +692,7 @@ export const getSnapController = async (
 
 export const getSnapControllerWithEES = async (
   options = getSnapControllerOptions(),
-  service?: ReturnType<typeof getNodeEES>,
+  service?: AbstractExecutionService<unknown>,
   init = true,
 ) => {
   const _service =