Sparse read streams

index.js

@@ -221,14 +221,17 @@ class Hyperbee extends EventEmitter {
 
   async inflate(ptr, config) {
     if (!ptr.value) {
-      await inflate(ptr, config)
+      const value = await inflate(ptr, config)
+      if (!value) return null
     }
+
     this.bump(ptr)
     return ptr.value
   }
 
   async finalizeKeyPointer(key, config) {
     const value = key.value || (await inflateValue(key, config))
+    if (value === null && !config.wait) return null
 
     return {
       core: key.context.getCore(key.core),
@@ -288,6 +291,7 @@ class Hyperbee extends EventEmitter {
 
     while (true) {
       const v = ptr.value ? this.bump(ptr) : await this.inflate(ptr, config)
+      if (!v) return null
 
       let s = 0
       let e = v.keys.length
@@ -297,6 +301,12 @@ class Hyperbee extends EventEmitter {
         const mid = (s + e) >> 1
         const m = v.keys.get(mid)
 
+        if (!m) {
+          if (s === mid) s = mid + 1
+          else e = mid
+          continue
+        }
+
         c = b4a.compare(key, m.key)
 
         if (c === 0) return this.finalizeKeyPointer(m, config)
@@ -309,6 +319,7 @@ class Hyperbee extends EventEmitter {
 
       const i = c < 0 ? e : s
       ptr = v.children.get(i)
+      if (!ptr) return null
     }
   }
 }

lib/context.js

@@ -1,6 +1,5 @@
 const b4a = require('b4a')
 const ScopeLock = require('scope-lock')
-const { BLOCK_NOT_AVAILABLE } = require('hypercore-errors')
 const { TreeNodePointer } = require('./tree.js')
 const { decodeBlock, TYPE_LATEST } = require('./encoding.js')
 
@@ -163,11 +162,17 @@ class CoreContext {
   }
 
   async getBlock(seq, core, config) {
-    if (core !== 0 && core - 1 >= this.cores.length) await this.update(config)
+    if (core !== 0 && core - 1 >= this.cores.length) {
+      if (!config.wait) return null
+      await this.update(config)
+    }
 
     const hc = this.getCore(core)
     const buffer = await hc.get(seq, config)
-    if (buffer === null) throw BLOCK_NOT_AVAILABLE()
+
+    if (buffer === null && !config.wait) {
+      return null
+    }
 
     if (config.trace !== null) config.trace(core, seq)
 
@@ -194,7 +199,10 @@ class CoreContext {
 
   async getContext(core, config) {
     if (core === 0) return this
-    if (core > this.cores.length) await this.update(config)
+    if (core > this.cores.length) {
+      if (!config.wait) return null
+      await this.update(config)
+    }
     if (core > this.cores.length) throw new Error('Bad core index: ' + core)
 
     const hex = b4a.toString(this.cores[core - 1].key, 'hex')

lib/inflate.js

@@ -9,6 +9,7 @@ exports.inflate = async function inflate(ptr, config) {
     ptr.context.getBlock(ptr.seq, ptr.core, config),
     ptr.context.getContext(ptr.core, config)
   ])
+  if (!config.wait && (!block || !context)) return null
 
   const tree = block.tree[ptr.offset]
 
@@ -32,6 +33,10 @@ exports.inflate = async function inflate(ptr, config) {
   return ptr.value
 }
 
+function missingDelta(d) {
+  return new DeltaOp(false, d.type, d.index, null)
+}
+
 function inflateKey(context, d, ptr, block, config) {
   if (d.type === OP_COHORT) return inflateKeyCohort(context, d, ptr, block, config)
   return inflateKeyDelta(context, d, ptr, block, config)
@@ -40,20 +45,23 @@ function inflateKey(context, d, ptr, block, config) {
 async function inflateKeyDelta(context, d, ptr, block, config) {
   const k = d.pointer
 
-  if (!k) return new DeltaOp(false, d.type, d.index, null)
+  if (!k) return missingDelta(d)
 
   const blk =
     k.seq === ptr.seq && k.core === 0 && ptr.core === 0
       ? block
       : await context.getBlock(k.seq, k.core, config)
 
+  if (!config.wait && !blk) return missingDelta(d)
+
   const bk = blk.keys[k.offset]
 
   let vp = null
 
   if (bk.valuePointer) {
     const p = bk.valuePointer
     const ctx = await context.getContext(k.core, config)
+    if (!config.wait && !ctx) return missingDelta(d)
     vp = new ValuePointer(ctx, p.core, p.seq, p.offset, p.split)
   }
 
@@ -69,6 +77,7 @@ exports.inflateValue = async function inflateValue(key, config) {
 
   if (ptr.split === 0) {
     const block = await ptr.context.getBlock(ptr.seq, ptr.core, config)
+    if (!config.wait && !block) return null
     return block.values[ptr.offset]
   }
 
@@ -77,6 +86,10 @@ exports.inflateValue = async function inflateValue(key, config) {
     blockPromises[i] = ptr.context.getBlock(ptr.seq - ptr.split + i, ptr.core, config)
   }
   const blocks = await Promise.all(blockPromises)
+
+  // if any block is missing, treat the whole value as missing.
+  if (!config.wait && blocks.includes(null)) return null
+
   const splitValue = new Array(blockPromises.length)
   for (let i = 0; i < splitValue.length - 1; i++) {
     splitValue[i] = blocks[i].values[0]
@@ -93,9 +106,12 @@ async function inflateKeyCohort(context, d, ptr, block, config) {
       ? block
       : await context.getBlock(co.seq, co.core, config)
 
+  if (!config.wait && !blk) return missingDelta(d)
+
   const cohort = blk.cohorts[co.offset]
   const promises = new Array(cohort.length)
   const ctx = await context.getContext(co.core, config)
+  if (!config.wait && !ctx) return missingDelta(d)
 
   for (let i = 0; i < cohort.length; i++) {
     const p = cohort[i]
@@ -109,7 +125,11 @@ async function inflateKeyCohort(context, d, ptr, block, config) {
 
 async function inflateChild(context, d, ptr, block, config) {
   if (d.type === OP_COHORT) return inflateChildCohort(context, d, ptr, block, config)
-  if (d.pointer && !context.hasCore(d.pointer.core)) await context.update(config)
+  const missingCore = d.pointer && !context.hasCore(d.pointer.core)
+
+  if (!config.wait && missingCore) return missingDelta(d)
+  else if (missingCore) await context.update(config)
+
   return inflateChildDelta(context, d, ptr, block, config)
 }
 
@@ -127,13 +147,23 @@ async function inflateChildCohort(context, d, ptr, block, config) {
       ? block
       : await context.getBlock(co.seq, co.core, config)
 
+  if (!config.wait && !blk) return missingDelta(d)
+
   const cohort = blk.cohorts[co.offset]
   const deltas = new Array(cohort.length)
   const ctx = await context.getContext(co.core, config)
+  if (!config.wait && !ctx) return missingDelta(d)
 
   for (let i = 0; i < cohort.length; i++) {
     const c = cohort[i]
-    if (c.pointer && !ctx.hasCore(c.pointer.core)) await ctx.update(config)
+    if (!config.wait) {
+      if (c.pointer && !ctx.hasCore(c.pointer.core)) {
+        deltas[i] = missingDelta(c)
+        continue
+      }
+    } else if (!ctx.hasCore(c.pointer.core)) {
+      await ctx.update(config)
+    }
     deltas[i] = inflateChildDelta(ctx, c, co, blk, config)
   }
 

lib/ranges.js

@@ -25,7 +25,7 @@ class RangeIterator {
     this.end = lte || lt || null
     this.inclusive = !!(reverse ? lte : gte)
     this.compare = (reverse ? gte : lte) ? 0 : -1
-    this.prefetch = prefetch
+    this.prefetch = prefetch && this.config.wait
     this.prefetching = null
   }
 
@@ -46,13 +46,20 @@ class RangeIterator {
       const v = top.node.value
         ? this.tree.bump(top.node)
         : await this.tree.inflate(top.node, this.config)
+      if (!v) {
+        this.stack.pop()
+        if (this.stack.length === 0) break
+        continue
+      }
 
       for (let i = 0; i < v.keys.length; i++) {
         const j = this.reverse ? v.keys.length - 1 - i : i
+        const key = v.keys.get(j)
 
-        const c = this.reverse
-          ? b4a.compare(v.keys.get(j).key, this.end)
-          : b4a.compare(this.start, v.keys.get(j).key)
+        // skip non-local keys
+        if (!this.config.wait && !key) continue
+
+        const c = this.reverse ? b4a.compare(key.key, this.end) : b4a.compare(this.start, key.key)
 
         if (c < 0) break
         top.offset = 2 * i + 1 + (c === 0 ? offset : 1)
@@ -64,19 +71,32 @@ class RangeIterator {
       if (!child || k >= v.children.length) break
 
       const j = this.reverse ? v.children.length - 1 - k : k
+      const node = v.children.get(j)
 
-      this.stack.push({
-        node: v.children.get(j),
-        offset: 0
-      })
+      if (this.config.wait || node) {
+        this.stack.push({ node, offset: 0 })
+      }
 
       top.offset++
+      if (!this.config.wait && !node) break
     }
   }
 
   async next() {
-    const key = await this.nextKey()
-    return key === null ? key : this.tree.finalizeKeyPointer(key, this.config)
+    let key = await this.nextKey()
+    if (key === null) return null
+
+    if (this.config.wait) {
+      return this.tree.finalizeKeyPointer(key, this.config)
+    }
+
+    // for local read-stream, iterate until we find a non null entry or run out of keys
+    while (true) {
+      const entry = await this.tree.finalizeKeyPointer(key, this.config)
+      if (entry !== null) return entry
+      key = await this.nextKey()
+      if (key === null) return null
+    }
   }
 
   async nextKey() {
@@ -85,6 +105,7 @@ class RangeIterator {
       const v = top.node.value
         ? this.tree.bump(top.node)
         : await this.tree.inflate(top.node, this.config)
+      if (!v) continue
 
       const offset = top.offset++
       const child = (offset & 1) === 0
@@ -94,16 +115,23 @@ class RangeIterator {
         this.stack.push(top)
         if (k < v.children.length) {
           const j = this.reverse ? v.children.length - 1 - k : k
-          this.stack.push({ node: v.children.get(j), offset: 0 })
+          const node = v.children.get(j)
+          if (this.config.wait || node) {
+            this.stack.push({ node, offset: 0 })
+          }
         }
         continue
       }
 
       if (k < v.keys.length) {
         const j = this.reverse ? v.keys.length - 1 - k : k
-
         const data = v.keys.get(j)
 
+        if (!this.config.wait && !data) {
+          this.stack.push(top)
+          continue
+        }
+
         const c = this.reverse
           ? this.start
             ? b4a.compare(this.start, data.key)

lib/session-config.js

@@ -27,6 +27,7 @@ class SessionConfig {
       wait = this.wait,
       trace = this.trace
     } = opts
+
     return this.sub(activeRequests, timeout, wait, trace)
   }