Fork handling

// WORKAROUND — see explanation above the tab
function pipeToIterator<T>(
  source: { pipeTo(t: ReturnType<typeof createTarget<T>>): Promise<void> },
  initialCursor: BlockCursor | undefined,
  onFork: (previousBlocks: BlockCursor[]) => Promise<BlockCursor | null>
): AsyncIterableIterator<{ data: T; ctx: any }> {

  type Slot =
    | { k: 'batch'; v: { data: T; ctx: any } }
    | { k: 'end' }
    | { k: 'error'; err: unknown }

  const queue: Slot[] = []
  let consumerWake: (() => void) | null = null
  let producerAck:  (() => void) | null = null
  const wake = () => { consumerWake?.(); consumerWake = null }

  ;(source.pipeTo as any)(createTarget({
    write: async ({ read }: any) => {
      for await (const batch of read(initialCursor)) {
        queue.push({ k: 'batch', v: batch })
        wake()
        await new Promise<void>(r => { producerAck = r })
      }
      queue.push({ k: 'end' })
      wake()
    },
    fork: onFork,
  })).catch((err: unknown) => { queue.push({ k: 'error', err }); wake() })

  return {
    async next(): Promise<IteratorResult<{ data: T; ctx: any }>> {
      if (!queue.length) await new Promise<void>(r => { consumerWake = r })
      const slot = queue.shift()!
      if (slot.k === 'end')   return { done: true,  value: undefined as any }
      if (slot.k === 'error') throw slot.err
      producerAck?.(); producerAck = null
      return { done: false, value: slot.v }
    },
    [Symbol.asyncIterator]() { return this },
  }
}

function findRollbackIndex(chainA: BlockCursor[], chainB: BlockCursor[]): number {
  let aIndex = 0, bIndex = 0, lastCommonIndex = -1
  while (aIndex < chainA.length && bIndex < chainB.length) {
    const a = chainA[aIndex], b = chainB[bIndex]
    if (a.number < b.number) { aIndex++; continue }
    if (a.number > b.number) { bIndex++; continue }
    if (a.hash !== b.hash) return lastCommonIndex
    lastCommonIndex = aIndex; aIndex++; bIndex++
  }
  return lastCommonIndex
}