Issue: fns.shift(...) is not a function error with libp2p v3

Description

OrbitDB sync fails with it-pipe error when using libp2p v3. The error occurs in sync.js when piping streams during peer synchronization.

Error

TypeError: fns.shift(...) is not a function
    at rawPipe (index.ts:297:20)
    at pipe (index.ts:291:10)
    at task (sync.js:198:17)

Affected Code

File: src/sync.js (in @orbitdb/core package)
Line: 198

const stream = await libp2p.dialProtocol(remotePeer, headsSyncAddress, { signal })
await pipe(sendHeads, stream, receiveHeads(peerId))

Versions

• @orbitdb/core: 3.0.2
• libp2p: 3.0.2 (v3)
• it-pipe: 3.0.1
• helia: 5.5.1

Root Cause

The issue is in how sendHeads is used as the first argument to pipe().

sendHeads is defined as a transform function:

const sendHeads = (source) => {
  return (async function * () {
    const heads = await log.heads()
    for await (const { hash } of heads) {
      const bytes = await log.storage.get(hash)
      yield bytes
    }
  })()
}

In it-pipe, the first argument should be:

• An async iterable (like stream from libp2p v3, which implements AsyncIterable<Uint8Array>)
• A function that returns an async iterable when called with no arguments

However, sendHeads is a function that expects a source parameter and returns an async generator. When passed directly to pipe() as the first argument, it-pipe's rawPipe function tries to call sendHeads() without arguments, but sendHeads expects a source parameter, causing the error.

Note: libp2p v3's Stream interface correctly implements AsyncIterable<Uint8Array | Uint8ArrayList> (see @libp2p/interface/src/message-stream.ts:79), so the stream itself is compatible with it-pipe. The issue is specifically with how sendHeads is being used.

Expected Behavior

Sync should work correctly with libp2p v3 streams. The sendHeads function should either:

1. Be called to return the async generator: pipe(sendHeads(), stream, receiveHeads(peerId))
2. Or be restructured to work as a source function for it-pipe

Additional Context

• Line 174 (pipe(stream, receiveHeads(peerId), sendHeads, stream)) works correctly because stream is the source and sendHeads is used as a transform
• Line 198 fails because sendHeads is incorrectly used as the source
• This appears to be a bug in OrbitDB's sync implementation, not a libp2p v3 compatibility issue