Function loadSubset

packages/electric-db-collection/src/electric.ts:443–537

(opts: LoadSubsetOptions)

  }

  const loadSubset = async (opts: LoadSubsetOptions) => {
    if (isBufferingInitialSync()) {
      const snapshotParams = compileSQL<T>(opts, compileOptions)
      try {
        const { data: rows } = await stream.fetchSnapshot(snapshotParams)

        if (!isBufferingInitialSync()) {
          debug(`${logPrefix}Ignoring snapshot - sync completed while fetching`)
          return
        }

        if (rows.length > 0) {
          begin()
          for (const row of rows) {
            write({
              type: `insert`,
              value: row.value,
              metadata: { ...row.headers },
            })
          }
          commit()
          debug(`${logPrefix}Applied snapshot with ${rows.length} rows`)
        }
      } catch (error) {
        if (handleSnapshotError(error, `fetchSnapshot`)) {
          return
        }
        throw error
      }
      return
    }

    if (syncMode === `progressive`) {
      return
    }

    const { cursor, where, orderBy, limit } = opts

    // When the stream is already up-to-date, it may be in a long-poll wait.
    // Forcing a disconnect-and-refresh ensures requestSnapshot gets a response
    // from a fresh server round-trip rather than waiting for the current poll to end.
    // If the refresh fails (e.g., PauseLock held during subscriber processing in
    // join pipelines), we fall through to requestSnapshot which still works.
    if (stream.isUpToDate) {
      try {
        await stream.forceDisconnectAndRefresh()
      } catch (error) {
        if (handleSnapshotError(error, `forceDisconnectAndRefresh`)) {
          return
        }
        debug(
          `${logPrefix}forceDisconnectAndRefresh failed, proceeding to requestSnapshot: %o`,
          error,
        )
      }
    }

    try {