RxDB集成Supabase时pullHandler无限重复调用问题排查

1. 实时订阅的RESYNC信号重复触发 #

在replication.ts的Supabase订阅逻辑中，每当订阅状态变为SUBSCRIBED时，就会向pullStream$发送"RESYNC"信号：

supabaseClient
  .from(DbTables.Block)
  .on("*", (payload) => { /* ... */ })
  .subscribe((status: string) => {
    if (status === "SUBSCRIBED") {
      pullStream$.next("RESYNC");
    }
  });

RxDB收到RESYNC信号后，会强制触发全量同步（忽略当前checkpoint，从头拉取数据）。如果Supabase实时订阅因网络波动、重连等原因多次触发SUBSCRIBED状态，就会导致pullHandler被无限调用。

解决方式：添加标志位，确保RESYNC仅在首次订阅成功时发送一次：

let hasSentInitialResync = false;

supabaseClient
  .from(DbTables.Block)
  .on("*", (payload) => { /* ... */ })
  .subscribe((status: string) => {
    if (status === "SUBSCRIBED" && !hasSentInitialResync) {
      hasSentInitialResync = true;
      pullStream$.next("RESYNC");
    }
  });

2. Checkpoint更新逻辑存在缺陷 #

在blockPullHandler中，拉取空数据时直接返回原lastCheckpoint：

checkpoint:
  docs.length === 0
    ? lastCheckpoint
    : {
        id: lastOfArray(docs).id,
        updated: lastOfArray(docs).updated,
      },

• 首次同步时若没有数据，返回的checkpoint为undefined，RxDB会认为同步未完成，重复调用pullHandler；
• 使用gt("updated", minTimestamp)作为查询条件，会漏掉updated时间等于minTimestamp的文档，导致RxDB始终认为存在未同步内容。

解决方式：

• 首次同步无数据时，返回初始checkpoint（如{ updated: Date.now() }），避免返回undefined；
• 将查询条件从gt改为gte，确保不遗漏时间戳相同的文档：

const { data, error } = await supabaseClient
  .from(DbTables.Block)
  .select()
  .gte("updated", minTimestamp)
  .order("updated", { ascending: true })
  .limit(batchSize);

3. 额外验证步骤 #

• 查看控制台Pulling data日志，确认每次调用的lastCheckpoint是否变化，若始终相同，说明checkpoint更新逻辑异常；
• 在Supabase订阅的subscribe回调中打印完整状态细节，确认是否存在SUBSCRIBED状态的重复触发。

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