Hasura GraphQL数据库写入后查询引发竞态条件问题求助

问题场景 #

遍历事件数组处理时，同一用户的两个事件（支付/非支付类型）会触发竞态：第一个事件的用户写入操作未完成，第二个事件的查询就判定用户不存在，进而尝试重复写入，最终抛出user_pkey主键约束错误。当前代码采用"先查询用户是否存在，不存在则写入"的逻辑，但并行执行时会出现冲突。

问题代码示例 #

for (let event of events) {
    if (event.type == payment) {
        // 检查用户是否存在
        let { data } = await client.query({
          query: user_query,
          variables: {id: event.user}
        })
        if (!data.User_by_pk) {
          await client.mutate({
            mutation: user_mutation,
            variables: {id: event.user}
          })
        } 
    } else {
        // 同样的检查逻辑
        let { data } = await client.query({
          query: user_query,
          variables: {id: event.user}
        })
        if (!data.User_by_pk) {
          await client.mutate({
            mutation: user_mutation,
            variables: {id: event.user}
          })
        }
    }
}

最优解决方案 #

1. 使用Hasura的upsert突变（推荐） #

Hasura支持原子性的upsert操作（插入或忽略/更新），直接跳过"先查后写"的步骤，让数据库层面保证数据一致性。修改你的用户突变逻辑为upsert：

# 定义upsert突变
mutation UpsertUser($id: uuid!) {
  insert_user(
    objects: {id: $id},
    on_conflict: {constraint: user_pkey, update_columns: []} # 冲突时不执行任何更新
  ) {
    affected_rows
  }
}

之后代码中直接调用该突变，无需提前查询：

for (let event of events) {
    // 不管事件类型，直接执行upsert
    await client.mutate({
      mutation: upsert_user_mutation,
      variables: {id: event.user}
    })
    // 后续事件处理逻辑...
}

数据库会自动处理：用户存在则无操作，不存在则插入，全程原子性，彻底规避竞态。

2. 对用户ID去重后串行处理 #

先提取事件中所有唯一用户ID，逐个处理确保同一用户只被创建一次，再处理后续事件：

// 提取所有唯一用户ID
const uniqueUserIds = [...new Set(events.map(event => event.user))];

// 串行处理每个用户的创建逻辑
for (let userId of uniqueUserIds) {
    let { data } = await client.query({
      query: user_query,
      variables: {id: userId}
    })
    if (!data.User_by_pk) {
      await client.mutate({
        mutation: user_mutation,
        variables: {id: userId}
      })
    }
}

// 再遍历事件执行后续处理
for (let event of events) {
    // 事件处理逻辑...
}

3. 用缓存标记已处理用户 #

在循环中维护一个缓存集合，记录已经完成查写操作的用户，避免重复执行：

const processedUsers = new Set();

for (let event of events) {
    const userId = event.user;
    if (processedUsers.has(userId)) {
        // 已处理过，直接跳过用户创建步骤
        continue;
    }

    let { data } = await client.query({
      query: user_query,
      variables: {id: userId}
    })
    if (!data.User_by_pk) {
      await client.mutate({
        mutation: user_mutation,
        variables: {id: userId}
      })
    }
    // 标记该用户已处理
    processedUsers.add(userId);
}

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