fix: Destroying execution environment streams should emit events (#3074)

Previously we would remove event listeners when calling `destroy`. This
meant that consumers of our streams would not be notified that the
stream was closed causing issues like `Cannot call write after a stream
was destroyed`.

Instead, we want consumers to be notified when a stream is torn down.
One issue however is that it introduces new issues where some code paths
throw `Premature close`. This seems to be an underlying problem in the
MetaMask stream architecture and the error message is [ignored on the
extension
client](https://github.com/MetaMask/metamask-extension/blob/bb90c361992d0381ba1852143d6c08d2ccb2fce1/app/scripts/lib/stream-utils.js#L14-L16).
We do the same here.

This will eventually also allow us to remove `closeConnections` from the
SnapController constructor.

Author: FrederikBolding

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

@@ -193,8 +193,9 @@ export abstract class AbstractExecutionService<WorkerType>
 
     Object.values(job.streams).forEach((stream) => {
       try {
-        !stream.destroyed && stream.destroy();
-        stream.removeAllListeners();
+        if (!stream.destroyed) {
+          stream.destroy();
+        }
       } catch (error) {
         logError('Error while destroying stream', error);
       }
@@ -226,7 +227,7 @@ export abstract class AbstractExecutionService<WorkerType>
       streams.command,
       jsonRpcConnection.stream,
       (error) => {
-        if (error) {
+        if (error && !error.message?.match('Premature close')) {
           logError(`Command stream failure.`, error);
         }
       },
@@ -496,7 +497,7 @@ export function setupMultiplex(
 ): ObjectMultiplex {
   const mux = new ObjectMultiplex();
   pipeline(connectionStream, mux, connectionStream, (error) => {
-    if (error) {
+    if (error && !error.message?.match('Premature close')) {
       logError(`"${streamName}" stream failure.`, error);
     }
   });

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

@@ -1661,7 +1661,7 @@ describe('SnapController', () => {
       engine.push(middleware);
       const providerStream = createEngineStream({ engine });
       pipeline(stream, providerStream, stream, (error) => {
-        if (error) {
+        if (error && !error.message?.match('Premature close')) {
           logError(`Provider stream failure.`, error);
         }
       });
@@ -1736,7 +1736,7 @@ describe('SnapController', () => {
       engine.push(middleware);
       const providerStream = createEngineStream({ engine });
       pipeline(stream, providerStream, stream, (error) => {
-        if (error) {
+        if (error && !error.message?.match('Premature close')) {
           logError(`Provider stream failure.`, error);
         }
       });

packages/snaps-controllers/src/test-utils/execution-environment.ts

@@ -56,7 +56,7 @@ export const getNodeEES = (
         });
         const providerStream = createEngineStream({ engine });
         pipeline(stream, providerStream, stream, (error) => {
-          if (error) {
+          if (error && !error.message?.match('Premature close')) {
             logError(`Provider stream failure.`, error);
           }
         });

packages/snaps-controllers/src/test-utils/service.ts

@@ -44,7 +44,7 @@ export const createService = <
       });
       const providerStream = createEngineStream({ engine });
       pipeline(stream, providerStream, stream, (error) => {
-        if (error) {
+        if (error && !error.message?.match('Premature close')) {
           logError(`Provider stream failure.`, error);
         }
       });

packages/snaps-execution-environments/src/common/BaseSnapExecutor.ts

@@ -403,7 +403,7 @@ export class BaseSnapExecutor {
 
     const multiplex = new ObjectMultiplex();
     pipeline(this.rpcStream, multiplex, this.rpcStream, (error) => {
-      if (error) {
+      if (error && !error.message?.match('Premature close')) {
         logError(`Provider stream failure.`, error);
       }
     });

packages/snaps-simulation/src/simulation.ts

@@ -393,8 +393,8 @@ export async function installSnap<
 
       // Error function is difficult to test, so we ignore it.
       /* istanbul ignore next 2 */
-      pipeline(stream, providerStream, stream, (error: unknown) => {
-        if (error) {
+      pipeline(stream, providerStream, stream, (error) => {
+        if (error && !error.message?.match('Premature close')) {
           logError(`Provider stream failure.`, error);
         }
       });