为 Playwright E2E 测试构建一个处理数据库读写分离延迟的 Jest 环境

端到端 (E2E) 测试的稳定性是衡量其价值的核心指标。在一个采用数据库读写分离架构的系统中，一个常见的痛点是测试的随机性失败，其根源往往指向主从数据库之间的复制延迟。一个典型的场景是：测试用例通过 Playwright 创建了一个新用户，紧接着尝试用该用户的凭据登录，但由于读取操作被路由到尚未同步新用户数据的从库，导致断言失败。

在真实项目中，依赖 page.waitForTimeout(2000) 这类硬编码的等待，是一种脆弱且低效的反模式。它不仅会显著拖慢整个测试套件的执行速度，而且无法保证在不同负载下的可靠性。我们需要的是一个确定性的、与架构耦合的解决方案，能够让测试代码感知到数据复制的状态。

这里的核心挑战在于，测试框架（Jest + Playwright）本身对后端的数据架构一无所知。我们的目标，就是构建一座桥梁，将后端的复制状态透明地暴露给前端的测试执行器。我们将通过实现一个自定义的 Jest 环境来解决这个问题。

初始的困境：一个典型的失败案例

让我们从一个具体的失败测试开始。假设我们有以下 Playwright 测试代码，使用 Jest作为测试运行器：

// tests/auth.e2e.spec.ts
import { test, expect, Page } from '@playwright/test';
import { v4 as uuidv4 } from 'uuid';

test.describe('Authentication Flow with Read-Write Splitting', () => {
  let page: Page;
  const uniqueEmail = `user-${uuidv4()}@example.com`;
  const password = 'password123';

  test.beforeAll(async ({ browser }) => {
    page = await browser.newPage();
  });

  test('should fail due to replication lag', async () => {
    // 步骤 1: 注册一个新用户 (这是一个写操作)
    // 这个请求会被路由到主数据库
    await page.goto('/register');
    await page.fill('input[name="email"]', uniqueEmail);
    await page.fill('input[name="password"]', password);
    await page.click('button[type="submit"]');
    await expect(page.locator('h1')).toHaveText('Registration Successful');

    // 步骤 2: 立即尝试用新凭据登录 (这是一个读操作)
    // 这个请求很可能被路由到尚未同步数据的从数据库
    await page.goto('/login');
    await page.fill('input[name="email"]', uniqueEmail);
    await page.fill('input[name="password"]', password);
    await page.click('button[type="submit"]');

    // 断言: 这里会随机性失败
    // 如果从库延迟，后端服务会找不到用户，返回错误信息
    await expect(page.locator('.dashboard-welcome')).toHaveText(`Welcome, ${uniqueEmail}`);
    // 实际得到的结果可能是 "Invalid credentials" 错误提示
  });

  test.afterAll(async () => {
    await page.close();
  });
});

这个测试在本地开发环境（通常没有读写分离）下可能永远是成功的，但在预生产或生产环境中，它的失败率会非常高。这正是我们需要解决的核心问题。

方案设计：自定义 Jest 环境与一致性协调器

直接修改应用代码来适应测试是一种侵入性很强的做法，应该避免。更好的方式是从测试基础设施层面入手。Jest 允许我们通过 testEnvironment 配置项提供一个自定义的测试环境类。这个类可以在测试套件的生命周期中执行特定逻辑，这为我们实现一致性等待提供了完美的钩子。

我们的方案由三部分组成：

1. 自定义 Jest 环境 (ReplicationAwareJestEnvironment): 继承自 jest-environment-node，它将负责管理一个与后端通信的协调器实例。
2. 一致性协调器 (ConsistencyCoordinator): 一个在测试环境中运行的轻量级客户端，负责在“写”操作后通知后端，并在“读”操作前向后端查询复制状态，阻塞测试直到数据同步完成。
3. 后端信标 API (/_test/consistency): 在应用后端实现一组仅在测试环境下暴露的内部 API，用于接收“写”信号和报告主从同步状态。

下面的流程图清晰地展示了这个交互过程：

sequenceDiagram
    participant Test as Test Execution (Jest)
    participant Page as Playwright Page
    participant App as Application Backend
    participant MasterDB as Master Database
    participant ReplicaDB as Replica Database

    Test->>Page: 触发注册操作 (e.g., page.click('register'))
    Page->>App: POST /register request
    App->>MasterDB: INSERT INTO users (...)
    MasterDB-->>App: Success
    App-->>Page: 200 OK, Registration successful
    Page-->>Test: 操作完成

    Note over Test: 注册操作(写)完成
    Test->>App: 调用协调器: POST /_test/consistency/signal-write
    Note over App: 记录当前 Master DB 的 LSN/Timestamp
    App-->>Test: 200 OK, Signal received

    Test->>Page: 触发登录操作 (e.g., page.click('login'))
    Note over Test: 在执行实际的读操作前，先确保一致性
    Test->>App: 调用协调器: GET /_test/consistency/wait-for-sync
    App->>ReplicaDB: 查询 Replica 当前的 LSN/Timestamp
    loop 直到 Replica LSN >= Master LSN
        App->>ReplicaDB: RECHECK LSN
        ReplicaDB-->>App: Current LSN
    end
    App-->>Test: 200 OK, Replica is synced

    Page->>App: POST /login request
    App->>ReplicaDB: SELECT * FROM users WHERE email=...
    ReplicaDB-->>App: User data found
    App-->>Page: 200 OK, Login successful
    Page-->>Test: 操作完成
    Test->>Test: 执行断言 (expect)

步骤化实现

1. 后端信标 API

首先，我们需要在应用后端（例如一个 Express.js 应用）中添加两个内部 API 端点。这些端点应该只在 NODE_ENV=test 时才被启用。

// src/server.ts (示意代码)
import express from 'express';
import { getMasterStatus, getReplicaStatus, compareStatuses } from './db-utils';

const app = express();
app.use(express.json());

// 模拟一个变量来存储主库的最新状态标识
// 在真实项目中，这应该是 LSN (Log Sequence Number) 或 GTID (Global Transaction ID)
let lastKnownMasterStatus: any = null;

if (process.env.NODE_ENV === 'test') {
  console.log('✅ Test-only consistency endpoints are enabled.');

  // 端点1: 接收来自测试的 "写操作完成" 信号
  app.post('/_test/consistency/signal-write', async (req, res) => {
    try {
      // 获取当前主数据库的最新事务位置
      lastKnownMasterStatus = await getMasterStatus();
      if (!lastKnownMasterStatus) {
        // 在真实项目中，获取状态失败应被严肃处理
        console.error('Failed to get master DB status.');
        return res.status(500).json({ error: 'Could not retrieve master status.' });
      }
      console.log(`[Consistency] Signaled write. Master status:`, lastKnownMasterStatus);
      res.status(200).json({ status: 'signaled', master: lastKnownMasterStatus });
    } catch (error) {
      console.error('Error in /signal-write:', error);
      res.status(500).json({ error: 'Internal server error' });
    }
  });

  // 端点2: 阻塞式等待，直到从库追上之前标记的主库状态
  app.get('/_test/consistency/wait-for-sync', async (req, res) => {
    if (!lastKnownMasterStatus) {
      // 如果没有先调用 signal-write，直接返回
      return res.status(200).json({ status: 'no_signal', message: 'No write was signaled.' });
    }

    const timeout = 5000; // 设置一个最大等待超时，防止测试永远卡住
    const interval = 100; // 轮询间隔
    const startTime = Date.now();

    try {
      while (Date.now() - startTime < timeout) {
        const currentReplicaStatus = await getReplicaStatus();
        console.log(`[Consistency] Checking sync... Master: ${lastKnownMasterStatus}, Replica: ${currentReplicaStatus}`);
        
        // compareStatuses 是一个特定于数据库的函数，用于比较主从位置
        if (compareStatuses(currentReplicaStatus, lastKnownMasterStatus) >= 0) {
          console.log(`[Consistency] Sync confirmed.`);
          lastKnownMasterStatus = null; // 重置状态
          return res.status(200).json({ status: 'synced' });
        }
        await new Promise(resolve => setTimeout(resolve, interval));
      }
      console.error(`[Consistency] Timeout waiting for replica to sync.`);
      res.status(408).json({ error: 'Timeout waiting for replica sync' });
    } catch (error) {
      console.error('Error in /wait-for-sync:', error);
      res.status(500).json({ error: 'Internal server error' });
    }
  });
}

// ... 其他应用路由

这里的 getMasterStatus, getReplicaStatus, compareStatuses 是数据库强相关的实现。例如，对于 PostgreSQL，它们可能涉及查询 pg_current_wal_lsn() 和 pg_last_wal_replay_lsn()。对于 MySQL，可能涉及 GTID。

2. 一致性协调器

这是一个简单的客户端，封装了对后端信标 API 的调用。我们将把它放在测试代码的工具目录下。

// tests/utils/consistencyCoordinator.ts
import fetch from 'node-fetch';

const BASE_URL = process.env.APP_URL || 'http://localhost:3000';

export class ConsistencyCoordinator {
  private apiUrl: string;

  constructor(baseUrl: string = BASE_URL) {
    this.apiUrl = baseUrl;
    console.log(`[Coordinator] Initialized with base URL: ${this.apiUrl}`);
  }

  /**
   * 通知后端一个写操作已经发生。
   * 这应该在任何可能导致复制延迟的数据库写操作之后立即调用。
   */
  async signalWrite(): Promise<void> {
    try {
      const response = await fetch(`${this.apiUrl}/_test/consistency/signal-write`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
      });
      if (!response.ok) {
        throw new Error(`Failed to signal write: ${response.statusText}`);
      }
      console.log('[Coordinator] Successfully signaled a write operation.');
    } catch (error) {
      console.error('[Coordinator] Error signaling write:', error);
      // 在CI环境中，我们可能希望测试直接失败而不是继续
      throw error;
    }
  }

  /**
   * 阻塞执行，直到后端确认从库数据与之前标记的主库状态同步。
   * 这应该在任何依赖于先前写操作结果的读操作之前调用。
   */
  async waitForSync(): Promise<void> {
    try {
      console.log('[Coordinator] Waiting for replica sync...');
      const response = await fetch(`${this.apiUrl}/_test/consistency/wait-for-sync`);
      if (!response.ok) {
        const body = await response.text();
        throw new Error(`Failed to wait for sync: ${response.statusText}. Body: ${body}`);
      }
      console.log('[Coordinator] Replica sync confirmed.');
    } catch (error) {
      console.error('[Coordinator] Error waiting for sync:', error);
      throw error;
    }
  }
}

3. 自定义 Jest 环境

现在是核心部分。我们创建自己的 Jest 环境，它会在全局作用域中注入一个 consistencyCoordinator 实例，以便所有测试文件都能访问到。

// tests/jest.environment.ts
import NodeEnvironment from 'jest-environment-node';
import type { JestEnvironmentConfig, EnvironmentContext } from '@jest/environment';
import { ConsistencyCoordinator } from './utils/consistencyCoordinator';

class ReplicationAwareJestEnvironment extends NodeEnvironment {
  constructor(config: JestEnvironmentConfig, context: EnvironmentContext) {
    super(config, context);

    // 将我们的协调器实例挂载到全局对象上
    // 这样，在 *.spec.ts 文件中就可以直接使用 `global.consistencyCoordinator`
    // 或者，如果你配置了 TypeScript 的 global 类型，可以直接使用 `consistencyCoordinator`
    this.global.consistencyCoordinator = new ConsistencyCoordinator();
  }

  async setup() {
    await super.setup();
    // 可以在这里执行每个测试套件开始前的全局设置
    console.log('[Jest Environment] Setup complete. Coordinator is available globally.');
  }

  async teardown() {
    // 可以在这里执行每个测试套件结束后的全局清理
    console.log('[Jest Environment] Teardown.');
    await super.teardown();
  }
}

export default ReplicationAwareJestEnvironment;

为了让 Jest 使用这个环境，我们需要更新 jest.config.js：

// jest.config.js
module.exports = {
  // ... other jest configurations
  testEnvironment: './tests/jest.environment.ts',
  // 如果使用 TypeScript，确保 jest-environment-node 的类型是可用的
};

同时，为了在测试代码中获得类型提示，可以创建一个类型声明文件：

// tests/global.d.ts
import { ConsistencyCoordinator } from './utils/consistencyCoordinator';

declare global {
  var consistencyCoordinator: ConsistencyCoordinator;
}

确保这个文件被 tsconfig.json 的 include 字段覆盖。

4. 重构测试用例

万事俱备，现在我们可以重构最初失败的测试用例，使其变得健壮。

// tests/auth.e2e.spec.ts
import { test, expect, Page } from '@playwright/test';
import { v4 as uuidv4 } from 'uuid';
// 确保 global.d.ts 被 tsconfig 识别
// 现在可以直接使用 consistencyCoordinator

test.describe('Authentication Flow with Read-Write Splitting', () => {
  let page: Page;
  const uniqueEmail = `user-${uuidv4()}@example.com`;
  const password = 'password123';

  test.beforeAll(async ({ browser }) => {
    page = await browser.newPage();
  });

  test('should succeed by handling replication lag', async () => {
    // 步骤 1: 注册一个新用户 (写操作)
    await page.goto('/register');
    await page.fill('input[name="email"]', uniqueEmail);
    await page.fill('input[name="password"]', password);
    await page.click('button[type="submit"]');
    await expect(page.locator('h1')).toHaveText('Registration Successful');

    // **关键步骤: 写操作完成后，立即发出信号**
    await global.consistencyCoordinator.signalWrite();

    // 步骤 2: 尝试登录 (读操作)
    await page.goto('/login');

    // **关键步骤: 读操作开始前，等待数据同步**
    await global.consistencyCoordinator.waitForSync();

    await page.fill('input[name="email"]', uniqueEmail);
    await page.fill('input[name="password"]', password);
    await page.click('button[type="submit"]');

    // 断言: 现在这个断言将是稳定和确定性的
    await expect(page.locator('.dashboard-welcome')).toHaveText(`Welcome, ${uniqueEmail}`);
  });

  test.afterAll(async () => {
    await page.close();
  });
});

通过显式调用 signalWrite 和 waitForSync，我们的测试逻辑现在与底层数据架构的异步性保持了同步。测试不再是盲目地假设数据立即可用，而是主动地去确认。

方案的局限性与未来展望

这个方案有效地解决了读写分离架构下 E2E 测试的稳定性问题，但它并非没有权衡。

当前的实现要求测试编写者手动在代码中插入 signalWrite 和 waitForSync 调用。这增加了心智负担，并且容易遗漏。一个潜在的优化方向是进一步封装 Playwright 的 page 对象。我们可以创建一个代理 page 对象，它自动拦截所有可能引起状态变更的 HTTP 请求（如 POST, PUT, DELETE, PATCH），并在内部调用 signalWrite。同样，它可以拦截后续的导航或数据获取请求，并自动注入 waitForSync 逻辑。这种方式对测试编写者来说几乎是透明的，但实现起来会更复杂，需要对 Playwright 的 API 有更深入的理解。

此外，该方案依赖于后端提供特定的测试 API。这是一种受控的技术债，需要在团队内部达成共识，并确保这些 API 不会暴露在生产环境中。

最后，该模式的适用边界是明确的：它主要针对那些因数据复制延迟而导致不稳定的、涉及状态变更的 E2E 测试。对于不改变状态的只读测试、单元测试或不涉及读写分离架构的系统，这套机制则是不必要的复杂化。