feat(messaging-streams): manually deepen all six stream utility modules

- chunker: framed seq chunks, reassemble, setChunkSize
- multiplex: closeAll, getStream, per-stream byte counters
- backpressure: low/high water marks, bufferedBytes
- tee: branchBytes, removeBranch, depth guard
- transform: pipeTo and outbound backpressure
- resume-token: labeled checkpoints, clearTokens, bytesFromOffset
- Update category README and chunker tests

Co-authored-by: Cursor <[email protected]>

messaging-streams/README.md

@@ -8,12 +8,12 @@ Local and composable byte-stream utilities: chunking, multiplexing, backpressure
 
 | Module | Protocol | Role |
 |--------|----------|------|
-| [hyper-p2p-stream-chunker](./hyper-p2p-stream-chunker/) | `stream-chunker/v1` | Fixed-size chunks, `pendingLength`, `reset` |
+| [hyper-p2p-stream-chunker](./hyper-p2p-stream-chunker/) | `stream-chunker/v1` | Framed seq chunks, `reassemble`, `setChunkSize` |
-| [hyper-p2p-stream-multiplex](./hyper-p2p-stream-multiplex/) | `stream-multiplex/v1` | `listOpenStreamIds`, `hasStream` |
+| [hyper-p2p-stream-multiplex](./hyper-p2p-stream-multiplex/) | `stream-multiplex/v1` | `closeAll`, `getStream`, backpressure on frames |
-| [hyper-p2p-stream-backpressure](./hyper-p2p-stream-backpressure/) | `stream-backpressure/v1` | `drain`, `isPaused`, high-water mark |
+| [hyper-p2p-stream-backpressure](./hyper-p2p-stream-backpressure/) | `stream-backpressure/v1` | Low/high water marks, `bufferedBytes`, `setWaterMarks` |
-| [hyper-p2p-stream-tee](./hyper-p2p-stream-tee/) | `stream-tee/v1` | `drainBranch`, `drainAll`, `listBranches` |
+| [hyper-p2p-stream-tee](./hyper-p2p-stream-tee/) | `stream-tee/v1` | `branchBytes`, `removeBranch`, max depth guard |
-| [hyper-p2p-stream-transform](./hyper-p2p-stream-transform/) | `stream-transform/v1` | `compose`, `flush`, `readAll` |
+| [hyper-p2p-stream-transform](./hyper-p2p-stream-transform/) | `stream-transform/v1` | `pipeTo`, backpressure, `compose` |
-| [hyper-p2p-stream-resume-token](./hyper-p2p-stream-resume-token/) | `stream-resume-token/v1` | `pruneBefore`, `latestCheckpoint`, `listTokens` |
+| [hyper-p2p-stream-resume-token](./hyper-p2p-stream-resume-token/) | `stream-resume-token/v1` | Labeled checkpoints, `clearTokens`, `bytesFromOffset` |
 
 ## Quick start
 

messaging-streams/hyper-p2p-stream-backpressure/index.js

@@ -8,10 +8,11 @@ class HyperP2PStreamBackpressure extends EventEmitter {
   constructor (opts = {}) {
     super()
     this.highWaterMark = opts.highWaterMark || 65536
+    this.lowWaterMark = opts.lowWaterMark || Math.floor(this.highWaterMark * 0.5)
     this._buffer = []
     this._bytes = 0
     this._paused = false
-    this._stats = { written: 0, dropped: 0, paused: 0 }
+    this._stats = { written: 0, dropped: 0, paused: 0, resumed: 0 }
   }
 
   write (chunk) {
@@ -42,6 +43,7 @@ class HyperP2PStreamBackpressure extends EventEmitter {
 
   resume () {
     this._paused = false
+    this._stats.resumed++
     this.emit('resume')
     return true
   }
@@ -50,13 +52,25 @@ class HyperP2PStreamBackpressure extends EventEmitter {
     const buf = this._buffer.shift()
     if (!buf) return null
     this._bytes -= buf.length || buf.byteLength || 0
-    if (this._bytes < this.highWaterMark) this._paused = false
+    if (this._paused && this._bytes <= this.lowWaterMark) {
+      this._paused = false
+      this._stats.resumed++
+      this.emit('resume')
+    }
     return buf
   }
 
-  pending () { return this._buffer.length }
+  pending () {
+    return this._buffer.length
+  }
 
-  isPaused () { return this._paused }
+  bufferedBytes () {
+    return this._bytes
+  }
+
+  isPaused () {
+    return this._paused
+  }
 
   drain () {
     const out = []
@@ -74,12 +88,32 @@ class HyperP2PStreamBackpressure extends EventEmitter {
     return this
   }
 
-  getStats () {
+  setWaterMarks (high, low) {
-    return { ...this._stats, bytes: this._bytes, paused: this._paused, protocol: PROTOCOL }
+    this.highWaterMark = Number(high)
+    this.lowWaterMark = Number(low)
+    if (this.lowWaterMark > this.highWaterMark) {
+      throw new Error('lowWaterMark must be <= highWaterMark')
+    }
+    return { high: this.highWaterMark, low: this.lowWaterMark }
   }
 
-  async ready () { return this }
+  getStats () {
-  async close () { this._buffer = []; this._bytes = 0 }
+    return {
+      ...this._stats,
+      bytes: this._bytes,
+      paused: this._paused,
+      protocol: PROTOCOL
+    }
+  }
+
+  async ready () {
+    return this
+  }
+
+  async close () {
+    this.clear()
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PStreamBackpressure, PROTOCOL }

messaging-streams/hyper-p2p-stream-chunker/index.js

@@ -9,7 +9,9 @@ class HyperP2PStreamChunker extends EventEmitter {
     super()
     this.chunkSize = opts.chunkSize || 4096
     this._pending = b4a.alloc(0)
-    this._stats = { chunks: 0, bytes: 0 }
+    this._seq = 0
+    this._parts = []
+    this._stats = { chunks: 0, bytes: 0, flushed: 0 }
   }
 
   push (data) {
@@ -20,17 +22,28 @@ class HyperP2PStreamChunker extends EventEmitter {
     while (this._pending.length >= this.chunkSize) {
       const slice = this._pending.subarray(0, this.chunkSize)
       this._pending = this._pending.subarray(this.chunkSize)
-      this._stats.chunks++
+      const framed = this._frame(slice)
-      emitted.push(slice)
+      emitted.push(framed)
-      this.emit('chunk', slice)
+      this._parts.push(framed.data)
+      this.emit('chunk', framed)
     }
     return emitted
   }
 
-  pendingLength () { return this._pending.length }
+  _frame (data) {
+    const seq = this._seq++
+    this._stats.chunks++
+    return { seq, data, size: data.length }
+  }
+
+  pendingLength () {
+    return this._pending.length
+  }
 
   reset () {
     this._pending = b4a.alloc(0)
+    this._seq = 0
+    this._parts = []
     return this
   }
 
@@ -38,17 +51,52 @@ class HyperP2PStreamChunker extends EventEmitter {
     if (!this._pending.length) return null
     const tail = this._pending
     this._pending = b4a.alloc(0)
-    this._stats.chunks++
+    const framed = this._frame(tail)
-    this.emit('chunk', tail)
+    this._parts.push(framed.data)
-    return tail
+    this._stats.flushed++
+    this.emit('chunk', framed)
+    return framed
+  }
+
+  reassemble () {
+    if (!this._parts.length) return b4a.alloc(0)
+    return b4a.concat(this._parts)
+  }
+
+  reassembleFrom (chunks) {
+    if (!Array.isArray(chunks)) throw new Error('chunks must be an array')
+    const bufs = chunks.map((c) => (c && c.data ? c.data : c))
+    return b4a.concat(bufs)
+  }
+
+  setChunkSize (n) {
+    const size = Number(n)
+    if (!Number.isFinite(size) || size < 1) throw new Error('chunkSize must be positive')
+    this.chunkSize = size
+    return this.chunkSize
+  }
+
+  chunkCount () {
+    return this._stats.chunks
   }
 
   getStats () {
-    return { ...this._stats, pending: this._pending.length, protocol: PROTOCOL }
+    return {
+      ...this._stats,
+      pending: this._pending.length,
+      chunkSize: this.chunkSize,
+      protocol: PROTOCOL
+    }
   }
 
-  async ready () { return this }
+  async ready () {
-  async close () { this._pending = b4a.alloc(0) }
+    return this
+  }
+
+  async close () {
+    this.reset()
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PStreamChunker, PROTOCOL }

messaging-streams/hyper-p2p-stream-chunker/test/test.js

@@ -8,7 +8,16 @@ test('split chunks', async (t) => {
   const m = new HyperP2PStreamChunker({ chunkSize: 4 })
   const parts = m.push(require('b4a').from('abcdefgh'))
   t.is(parts.length, 2)
-  t.is(parts[0].length, 4)
+  t.is(parts[0].data.length, 4)
+  t.is(parts[0].seq, 0)
+  await m.close()
+})
+
+test('reassemble', async (t) => {
+  const m = new HyperP2PStreamChunker({ chunkSize: 3 })
+  m.push(require('b4a').from('hello'))
+  m.flush()
+  t.is(m.reassemble().toString(), 'hello')
   await m.close()
 })
 

messaging-streams/hyper-p2p-stream-multiplex/index.js

@@ -1,7 +1,6 @@
 require('bare-process/global')
 const EventEmitter = require('bare-events')
 const b4a = require('b4a')
-const { assertNonEmpty } = require('../../_shared/lib/errors.js')
 
 const PROTOCOL = 'stream-multiplex/v1'
 
@@ -11,6 +10,8 @@ class StreamHandle {
     this.id = id
     this._listeners = []
     this.closed = false
+    this.bytesWritten = 0
+    this.bytesRead = 0
   }
 
   write (chunk) {
@@ -32,6 +33,7 @@ class StreamHandle {
   }
 
   _emitData (chunk, fin) {
+    if (chunk) this.bytesRead += chunk.length || chunk.byteLength || 0
     for (const fn of this._listeners) fn(chunk, fin)
     if (fin) this.closed = true
   }
@@ -43,7 +45,7 @@ class HyperP2PStreamMultiplex extends EventEmitter {
     this.highWaterMark = opts.highWaterMark || 65536
     this._streams = new Map()
     this._nextId = 1
-    this._stats = { streams: 0, frames: 0, bytes: 0 }
+    this._stats = { streams: 0, frames: 0, bytes: 0, rejected: 0 }
   }
 
   openStream (id = null) {
@@ -59,6 +61,8 @@ class HyperP2PStreamMultiplex extends EventEmitter {
   _sendFrame (streamId, chunk, fin) {
     const size = chunk ? (chunk.length || chunk.byteLength || 0) : 0
     if (this._stats.bytes + size > this.highWaterMark && !fin) {
+      this._stats.rejected++
+      this.emit('backpressure', { streamId, bytes: this._stats.bytes })
       return false
     }
     const frame = { type: 'frame', streamId, chunk, fin: !!fin }
@@ -66,7 +70,10 @@ class HyperP2PStreamMultiplex extends EventEmitter {
     this._stats.bytes += size
     this.emit('frame', frame)
     const local = this._streams.get(streamId)
-    if (local) local._emitData(chunk, fin)
+    if (local) {
+      if (chunk) local.bytesWritten += size
+      local._emitData(chunk, fin)
+    }
     return true
   }
 
@@ -79,9 +86,21 @@ class HyperP2PStreamMultiplex extends EventEmitter {
     return true
   }
 
-  listOpenStreamIds () { return [...this._streams.keys()] }
+  openStreamCount () {
+    return this._streams.size
+  }
 
-  hasStream (streamId) { return this._streams.has(String(streamId)) }
+  listOpenStreamIds () {
+    return [...this._streams.keys()]
+  }
+
+  hasStream (streamId) {
+    return this._streams.has(String(streamId))
+  }
+
+  getStream (streamId) {
+    return this._streams.get(String(streamId)) || null
+  }
 
   closeStream (streamId) {
     const h = this._streams.get(streamId)
@@ -91,6 +110,11 @@ class HyperP2PStreamMultiplex extends EventEmitter {
     return true
   }
 
+  closeAll () {
+    for (const id of [...this._streams.keys()]) this.closeStream(id)
+    return this._streams.size === 0
+  }
+
   getStats () {
     return {
       ...this._stats,
@@ -99,10 +123,12 @@ class HyperP2PStreamMultiplex extends EventEmitter {
     }
   }
 
-  async ready () { return this }
+  async ready () {
+    return this
+  }
 
   async close () {
-    for (const id of [...this._streams.keys()]) this.closeStream(id)
+    this.closeAll()
     this.emit('closed')
   }
 }

messaging-streams/hyper-p2p-stream-resume-token/index.js

@@ -11,7 +11,7 @@ class HyperP2PStreamResumeToken extends EventEmitter {
     this._offset = 0
     this._tokens = new Map()
     this._nextToken = 1
-    this._stats = { bytes: 0, checkpoints: 0, resumes: 0 }
+    this._stats = { bytes: 0, checkpoints: 0, resumes: 0, pruned: 0 }
   }
 
   write (chunk) {
@@ -23,9 +23,9 @@ class HyperP2PStreamResumeToken extends EventEmitter {
     return true
   }
 
-  checkpoint () {
+  checkpoint (label = null) {
     const id = String(this._nextToken++)
-    const token = { id, offset: this._stats.bytes, at: Date.now() }
+    const token = { id, label, offset: this._stats.bytes, at: Date.now() }
     this._tokens.set(id, token)
     this._stats.checkpoints++
     this.emit('checkpoint', token)
@@ -52,7 +52,13 @@ class HyperP2PStreamResumeToken extends EventEmitter {
     return out
   }
 
-  listTokens () { return [...this._tokens.values()] }
+  bytesFromOffset () {
+    return Math.max(0, this._stats.bytes - this._offset)
+  }
+
+  listTokens () {
+    return [...this._tokens.values()]
+  }
 
   latestCheckpoint () {
     let best = null
@@ -78,16 +84,35 @@ class HyperP2PStreamResumeToken extends EventEmitter {
     }
     const removed = this._chunks.length - keep.length
     this._chunks = keep
+    this._stats.pruned += removed
     if (this._offset < cut) this._offset = cut
     return removed
   }
 
-  getStats () {
+  clearTokens () {
-    return { ...this._stats, offset: this._offset, protocol: PROTOCOL }
+    const n = this._tokens.size
+    this._tokens.clear()
+    return n
   }
 
-  async ready () { return this }
+  getStats () {
-  async close () { this._chunks = []; this._tokens.clear() }
+    return {
+      ...this._stats,
+      offset: this._offset,
+      tokens: this._tokens.size,
+      protocol: PROTOCOL
+    }
+  }
+
+  async ready () {
+    return this
+  }
+
+  async close () {
+    this._chunks = []
+    this._tokens.clear()
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PStreamResumeToken, PROTOCOL }

messaging-streams/hyper-p2p-stream-tee/index.js

@@ -8,24 +8,40 @@ class HyperP2PStreamTee extends EventEmitter {
   constructor (opts = {}) {
     super()
     this._branches = new Map()
-    this._stats = { written: 0, branches: 0 }
+    this._stats = { written: 0, branches: 0, dropped: 0 }
+    this.maxBranchDepth = opts.maxBranchDepth ?? 10000
   }
 
   addBranch (name) {
     if (!name) throw new Error('branch name required')
-    const branch = { name, chunks: [] }
+    const branch = { name, chunks: [], bytes: 0 }
     this._branches.set(name, branch)
     this._stats.branches++
-    return () => this._branches.delete(name)
+    this.emit('branch-added', { name })
+    return () => this.removeBranch(name)
+  }
+
+  removeBranch (name) {
+    return this._branches.delete(name)
   }
 
   write (chunk) {
     if (chunk == null) throw new Error('chunk required')
     const buf = typeof chunk === 'string' ? b4a.from(chunk) : chunk
-    for (const branch of this._branches.values()) branch.chunks.push(buf)
+    const size = buf.length || buf.byteLength || 0
+    let n = 0
+    for (const branch of this._branches.values()) {
+      if (branch.chunks.length >= this.maxBranchDepth) {
+        this._stats.dropped++
+        continue
+      }
+      branch.chunks.push(buf)
+      branch.bytes += size
+      n++
+    }
     this._stats.written++
     this.emit('data', buf)
-    return true
+    return n > 0
   }
 
   readBranch (name) {
@@ -38,14 +54,24 @@ class HyperP2PStreamTee extends EventEmitter {
     return b ? b.chunks.length : 0
   }
 
-  listBranches () { return [...this._branches.keys()] }
+  branchBytes (name) {
+    const b = this._branches.get(name)
+    return b ? b.bytes : 0
+  }
 
-  hasBranch (name) { return this._branches.has(name) }
+  listBranches () {
+    return [...this._branches.keys()]
+  }
+
+  hasBranch (name) {
+    return this._branches.has(name)
+  }
 
   drainBranch (name) {
     const b = this._branches.get(name)
     if (!b) return []
     const out = b.chunks.splice(0, b.chunks.length)
+    b.bytes = 0
     return out
   }
 
@@ -59,8 +85,14 @@ class HyperP2PStreamTee extends EventEmitter {
     return { ...this._stats, branchCount: this._branches.size, protocol: PROTOCOL }
   }
 
-  async ready () { return this }
+  async ready () {
-  async close () { this._branches.clear() }
+    return this
+  }
+
+  async close () {
+    this._branches.clear()
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PStreamTee, PROTOCOL }

messaging-streams/hyper-p2p-stream-transform/index.js

@@ -9,16 +9,22 @@ class HyperP2PStreamTransform extends EventEmitter {
     super()
     this._fn = opts.transform || null
     this._out = []
-    this._stats = { in: 0, out: 0 }
+    this._stats = { in: 0, out: 0, skipped: 0 }
+    this.highWaterMark = opts.highWaterMark ?? 1024
   }
 
   setTransform (fn) {
     if (typeof fn !== 'function') throw new Error('transform must be a function')
     this._fn = fn
+    return this
   }
 
   write (chunk) {
     if (chunk == null) throw new Error('chunk required')
+    if (this._out.length >= this.highWaterMark) {
+      this.emit('backpressure')
+      return false
+    }
     const buf = typeof chunk === 'string' ? b4a.from(chunk) : chunk
     this._stats.in++
     const result = this._fn ? this._fn(buf) : buf
@@ -27,18 +33,23 @@ class HyperP2PStreamTransform extends EventEmitter {
       this._out.push(out)
       this._stats.out++
       this.emit('data', out)
+    } else {
+      this._stats.skipped++
     }
     return true
   }
 
-  read () { return this._out.shift() || null }
+  read () {
-
+    return this._out.shift() || null
-  readAll () {
-    const out = this._out.splice(0, this._out.length)
-    return out
   }
 
-  flush () { return this.readAll() }
+  readAll () {
+    return this._out.splice(0, this._out.length)
+  }
+
+  flush () {
+    return this.readAll()
+  }
 
   compose (fn) {
     if (typeof fn !== 'function') throw new Error('fn must be a function')
@@ -51,14 +62,32 @@ class HyperP2PStreamTransform extends EventEmitter {
     return this
   }
 
-  pending () { return this._out.length }
+  pipeTo (target) {
-
+    if (!target || typeof target.write !== 'function') {
-  getStats () {
+      throw new Error('target must implement write()')
-    return { ...this._stats, protocol: PROTOCOL }
+    }
+    while (this._out.length) {
+      target.write(this.read())
+    }
+    return target
   }
 
-  async ready () { return this }
+  pending () {
-  async close () { this._out = [] }
+    return this._out.length
+  }
+
+  getStats () {
+    return { ...this._stats, pending: this._out.length, protocol: PROTOCOL }
+  }
+
+  async ready () {
+    return this
+  }
+
+  async close () {
+    this._out = []
+    this.emit('closed')
+  }
 }
 
 module.exports = { HyperP2PStreamTransform, PROTOCOL }