如何在Apollo Server中像处理查询/变更一样处理GraphQL订阅错误？

我尝试在Apollo Server中，用处理查询和变更时的formatError类似方式来处理GraphQL订阅的错误。以下是我的配置代码：

export default async () => {
  const typeDefs = readFileSync(path.resolve(import.meta.dirname, './schema.graphql'), 'utf8');
  const schema = makeExecutableSchema({typeDefs, resolvers});
  const app = express();
  const httpServer = http.createServer(app);
  const wsServer = new WebSocketServer({
    server: httpServer,
    path: '/subscriptions',
  });
  const pubsub = new PubSub();
  const serverCleanup = useServer({
    schema,
    context: async () => (Promise.resolve({
      pubsub: pubsub,
    })),
    onError: (ctx, msg, errors) => {
      console.log('onError', ctx, msg, errors); // 没有输出，服务器直接崩溃
      return [];
    },
  }, wsServer);

  const server = new ApolloServer({
    schema,
    plugins: [
      ApolloServerPluginDrainHttpServer({httpServer}),
      {
        async serverWillStart() {
          return Promise.resolve({
            async drainServer() {
              await serverCleanup.dispose();
            },
          });
        },
      },
    ],
    formatError: (formattedError) => {
      switch (formattedError.extensions?.code) {
        case GRAPHQL_USER_FRIENDLY_ERROR:
          return {message: formattedError.message};
        case ApolloServerErrorCode.INTERNAL_SERVER_ERROR:
          console.error(formattedError);
          return {message: 'An unexpected error occurred.'};
        default:
          return formattedError;
      }
    },
  });

  await server.start();

  app.use(
    '/graphql',
    cors(),
    express.json(),
    expressMiddleware(server, {
      context: async () => (Promise.resolve({
        pubsub: pubsub,
      })),
    }),
  );

  httpServer.listen(4000, () => {
    console.log('Server is ready!');
  });
}

依赖版本：

"@apollo/server": "^4.11.2",
"graphql-ws": "^5.16.0",
"ws": "^8.18.0"

我该如何捕获并处理Apollo Server中的GraphQL订阅错误（例如由一次性无效数据导致的错误），以防止服务器崩溃、记录错误或返回自定义错误消息，就像formatError处理查询和变更那样？

1. 修正onError与订阅全流程错误处理 #

当前onError未生效，是因为graphql-ws的错误触发分为订阅初始化和订阅数据推送两个阶段，需要分别处理：

• onError仅处理连接初始化、订阅请求解析阶段的错误
• 数据推送阶段的错误需要在迭代器内部捕获

修改useServer配置：

// 先抽离复用的错误格式化函数，和查询/变更保持一致
const formatErrorFn = (error) => {
  switch (error.extensions?.code) {
    case GRAPHQL_USER_FRIENDLY_ERROR:
      return { message: error.message };
    case ApolloServerErrorCode.INTERNAL_SERVER_ERROR:
      console.error('内部错误详情:', error);
      return { message: '发生了意外错误' };
    default:
      return error;
  }
};

const serverCleanup = useServer({
  schema,
  context: async () => ({ pubsub }),
  // 处理订阅初始化阶段的错误
  onError: (ctx, msg, errors) => {
    console.error('订阅初始化错误:', errors);
    // 向客户端发送格式化后的错误
    ctx.socket.send(JSON.stringify({
      id: msg.id,
      type: 'ERROR',
      payload: { errors: errors.map(formatErrorFn) },
    }));
  },
  // 拦截订阅执行，处理推送阶段的错误
  onSubscribe: async (ctx, msg) => {
    try {
      const contextValue = await ctx.context();
      const executionResult = await execute({
        schema,
        document: parse(msg.payload.query),
        contextValue,
        variableValues: msg.payload.variables,
        operationName: msg.payload.operationName,
      });

      // 如果是订阅操作，包装迭代器捕获推送错误
      if (executionResult[Symbol.asyncIterator]) {
        const iterator = executionResult[Symbol.asyncIterator]();
        return {
          [Symbol.asyncIterator]: async function* () {
            try {
              for await (const value of iterator) {
                yield value;
              }
            } catch (err) {
              console.error('订阅推送错误:', err);
              // 推送自定义错误给客户端
              ctx.socket.send(JSON.stringify({
                id: msg.id,
                type: 'ERROR',
                payload: { errors: [formatErrorFn(err)] },
              }));
              // 优雅关闭该订阅连接
              ctx.socket.close();
            }
          },
        };
      }
      return executionResult;
    } catch (err) {
      console.error('订阅执行错误:', err);
      return { errors: [formatErrorFn(err)] };
    }
  },
}, wsServer);

2. 全局兜底防止服务器崩溃 #

在Node.js层面添加全局错误监听，避免未捕获的异常导致整个服务崩溃：

// 捕获未处理的Promise拒绝
process.on('unhandledRejection', (reason) => {
  console.error('未处理的Promise拒绝:', reason);
});

// 捕获未捕获的同步异常
process.on('uncaughtException', (err) => {
  console.error('未捕获的异常:', err);
  // 优雅关闭服务器后退出
  httpServer.close(() => process.exit(1));
});

最佳实践 #

• 解析器内捕获错误：在订阅解析器内部用try/catch包裹业务逻辑，抛出带有明确code的自定义错误，避免错误冒泡到服务端顶层
• 统一错误格式：复用同一个错误格式化函数，确保查询、变更、订阅的错误返回格式完全一致
• 避免敏感信息泄露：所有对外返回的错误都要过滤掉堆栈、内部服务地址等敏感内容

内容的提问来源于stack exchange，提问作者Unlucky