Update many-level to be compatible with abstract-level 2.0 API

host.js

@@ -2,7 +2,8 @@
 
 const lpstream = require('@vweevers/length-prefixed-stream')
 const ModuleError = require('module-error')
-const { Duplex, finished } = require('readable-stream')
+const { Duplex, promises: readablePromises } = require('readable-stream')
+const { finished } = readablePromises
 const { input, output } = require('./tags')
 
 const rangeOptions = new Set(['gt', 'gte', 'lt', 'lte'])
@@ -23,7 +24,6 @@ const kBusy = Symbol('busy')
 const kPendingSeek = Symbol('pendingSeek')
 const kLimit = Symbol('limit')
 const kReadAhead = Symbol('readAhead')
-const noop = () => {}
 const limbo = Symbol('limbo')
 
 // TODO: make use of db.supports manifest
@@ -65,7 +65,7 @@ function createRpcStream (db, options, streamOptions) {
   const predel = options.predel
   const prebatch = options.prebatch
 
-  db.open({ passive: true }, ready)
+  db.open({ passive: true }).then(() => ready()).catch(ready)
 
   // TODO: send events to guest. Challenges:
   // - Need to know encodings or emit encoded data; current abstract-level events don't suffice
@@ -83,14 +83,23 @@ function createRpcStream (db, options, streamOptions) {
     if (err) return stream.destroy(err)
 
     const iterators = new Map()
+    const snapshots = new Map()
 
-    finished(stream, function () {
+    const cleanup = async () => {
+      await finished(stream).catch(() => null)
       for (const iterator of iterators.values()) {
         iterator.close()
       }
 
+      for (const snapshot of snapshots.values()) {
+        snapshot.close()
+      }
+
       iterators.clear()
-    })
+      snapshots.clear()
+    }
+    // Don't await
+    cleanup()
 
     decode.on('data', function (data) {
       if (!data.length) return
@@ -118,6 +127,10 @@ function createRpcStream (db, options, streamOptions) {
         case input.iteratorSeek: return oniteratorseek(req)
         case input.clear: return readonly ? onreadonly(req) : onclear(req)
         case input.getMany: return ongetmany(req)
+        case input.has: return onhas(req)
+        case input.hasMany: return onhasmany(req)
+        case input.snapshot: return onsnapshot(req)
+        case input.snapshotClose: return onsnapshotclose(req)
       }
     })
 
@@ -131,54 +144,108 @@ function createRpcStream (db, options, streamOptions) {
       encode.write(encodeMessage(msg, output.getManyCallback))
     }
 
-    function onput (req) {
-      preput(req.key, req.value, function (err) {
-        if (err) return callback(req.id, err)
-        db.put(req.key, req.value, encodingOptions, function (err) {
-          callback(req.id, err, null)
-        })
-      })
+    function hasCallback (id, err, value) {
+      const msg = { id, error: errorCode(err), value }
+      encode.write(encodeMessage(msg, output.hasCallback))
     }
 
-    function onget (req) {
-      db.get(req.key, encodingOptions, function (err, value) {
-        callback(req.id, err, value)
-      })
+    function hasManyCallback (id, err, values) {
+      const msg = { id, error: errorCode(err), values }
+      encode.write(encodeMessage(msg, output.hasManyCallback))
     }
 
-    function ongetmany (req) {
-      db.getMany(req.keys, encodingOptions, function (err, values) {
-        getManyCallback(req.id, err, values.map(value => ({ value })))
+    function snapshot (id) {
+      if (!id) return undefined
+
+      const snapshot = snapshots.get(id)
+      if (snapshot !== undefined) return snapshot
+
+      throw new ModuleError('Snapshot is not open', {
+        code: 'LEVEL_SNAPSHOT_NOT_OPEN'
       })
     }
 
-    function ondel (req) {
-      predel(req.key, function (err) {
-        if (err) return callback(req.id, err)
-        db.del(req.key, encodingOptions, function (err) {
-          callback(req.id, err)
-        })
-      })
+    async function onput (req) {
+      preput(req.key, req.value, function (err) { return callback(req.id, err) })
+      try {
+        await db.put(req.key, req.value, encodingOptions)
+      } catch (err) {
+        return callback(req.id, err, null)
+      }
+    }
+
+    async function onget (req) {
+      try {
+        const value = await db.get(req.key, withSnapshot(encodingOptions, snapshot(req.snapshot)))
+        return callback(req.id, null, value)
+      } catch (err) {
+        return callback(req.id, err, null)
+      }
+    }
+
+    async function ongetmany (req) {
+      try {
+        const values = await db.getMany(req.keys, withSnapshot(encodingOptions, snapshot(req.snapshot)))
+        return getManyCallback(req.id, null, values.map(value => ({ value })))
+      } catch (err) {
+        return getManyCallback(req.id, err, [])
+      }
+    }
+
+    async function onhas (req) {
+      try {
+        const value = await db.has(req.key, withSnapshot(encodingOptions, snapshot(req.snapshot)))
+        return hasCallback(req.id, null, value)
+      } catch (err) {
+        return hasCallback(req.id, err, false)
+      }
+    }
+
+    async function onhasmany (req) {
+      try {
+        const values = await db.hasMany(req.keys, withSnapshot(encodingOptions, snapshot(req.snapshot)))
+        return hasManyCallback(req.id, null, values)
+      } catch (err) {
+        return hasManyCallback(req.id, err, [])
+      }
+    }
+
+    async function ondel (req) {
+      predel(req.key, function (err) { return callback(req.id, err) })
+      try {
+        await db.del(req.key, encodingOptions)
+        return callback(req.id, null)
+      } catch (err) {
+        return callback(req.id, err)
+      }
     }
 
     function onreadonly (req) {
       callback(req.id, new ModuleError('Database is readonly', { code: 'LEVEL_READONLY' }))
     }
 
-    function onbatch (req) {
-      prebatch(req.ops, function (err) {
-        if (err) return callback(req.id, err)
-
-        db.batch(req.ops, encodingOptions, function (err) {
-          callback(req.id, err)
-        })
-      })
+    async function onbatch (req) {
+      prebatch(req.key, function (err) { return callback(req.id, err) })
+      try {
+        await db.batch(req.ops, encodingOptions)
+        return callback(req.id, null)
+      } catch (err) {
+        return callback(req.id, err)
+      }
     }
 
-    function oniterator ({ id, seq, options, consumed, bookmark, seek }) {
+    function oniterator ({ id, seq, options, consumed, bookmark, seek, snapshot: snapshotId }) {
       if (iterators.has(id)) return
 
-      const it = new Iterator(db, id, seq, options, consumed, encode)
+      try {
+        options = withSnapshot(options, snapshot(snapshotId))
+      } catch (err) {
+        const data = { id, error: errorCode(err), seq }
+        encode.write(encodeMessage(data, output.iteratorError))
+        return
+      }
+
+      const it = new ManyLevelHostIterator(db, id, seq, options, consumed, encode)
       iterators.set(id, it)
 
       if (seek) {
@@ -211,15 +278,38 @@ function createRpcStream (db, options, streamOptions) {
       if (it !== undefined) it.close()
     }
 
-    function onclear (req) {
-      db.clear(cleanRangeOptions(req.options), function (err) {
-        callback(req.id, err)
-      })
+    async function onclear (req) {
+      try {
+        const options = cleanRangeOptions(req.options)
+        await db.clear(withSnapshot(options, snapshot(options.snapshot)))
+        return callback(req.id, null)
+      } catch (err) {
+        return callback(req.id, err)
+      }
+    }
+
+    function onsnapshot (req) {
+      if (snapshots.has(req.id)) return
+
+      try {
+        snapshots.set(req.id, db.snapshot())
+      } catch (err) {
+        snapshots.set(req.id, {
+          close () {},
+          error: err
+        })
+      }
+    }
+
+    async function onsnapshotclose (req) {
+      const snapshot = snapshots.get(req.id)
+      snapshots.delete(req.id)
+      if (snapshot !== undefined) await snapshot.close()
     }
   }
 }
 
-class Iterator {
+class ManyLevelHostIterator {
   constructor (db, id, seq, options, consumed, encode) {
     options = cleanRangeOptions(options)
 
@@ -245,7 +335,7 @@ class Iterator {
     this.pendingAcks = 0
   }
 
-  next (first) {
+  async next (first) {
     if (this[kBusy] || this[kClosed]) return
     if (this[kEnded] || this.pendingAcks > 1) return
 
@@ -269,7 +359,12 @@ class Iterator {
     if (size <= 0) {
       process.nextTick(this[kHandleMany], null, [])
     } else {
-      this[kIterator].nextv(size, this[kHandleMany])
+      try {
+        const nextVal = await this[kIterator].nextv(size)
+        this[kHandleMany](null, nextVal)
+      } catch (err) {
+        this[kHandleMany](err, [])
+      }
     }
   }
 
@@ -341,10 +436,10 @@ class Iterator {
     }
   }
 
-  close () {
+  async close () {
     if (this[kClosed]) return
     this[kClosed] = true
-    this[kIterator].close(noop)
+    await this[kIterator].close()
   }
 }
 
@@ -399,3 +494,16 @@ function cleanRangeOptions (options) {
 
   return result
 }
+
+function withSnapshot (options, snapshot) {
+  if (snapshot === undefined) {
+    if (!options || !hasOwnProperty.call(options, 'snapshot')) return options
+
+    const result = { ...options }
+    delete result.snapshot
+    return result
+  }
+
+  if (snapshot.error) throw snapshot.error
+  return { ...options, snapshot }
+}

index.d.ts

@@ -2,7 +2,7 @@ import {
   AbstractLevel,
   AbstractDatabaseOptions,
   AbstractOpenOptions,
-  NodeCallback
+  AbstractGetOptions
 } from 'abstract-level'
 
 // Requires `npm install @types/readable-stream`.
@@ -25,8 +25,6 @@ export class ManyLevelGuest<KDefault = string, VDefault = string>
 
   open (): Promise<void>
   open (options: GuestOpenOptions): Promise<void>
-  open (callback: NodeCallback<void>): void
-  open (options: GuestOpenOptions, callback: NodeCallback<void>): void
 
   /**
    * Create a duplex guest stream to be piped into a host stream. Until that's done,
@@ -91,6 +89,12 @@ declare interface GuestDatabaseOptions<K, V> extends AbstractDatabaseOptions<K,
    */
   retry?: boolean
 
+  /**
+   * Enable {@link ManyLevelGuest.getSync}. If a function is provided, it will be called
+   * for remote reads after keys and options have been encoded by `abstract-level`.
+   */
+  getSync?: boolean | GuestGetSync
+
   /**
    * An {@link AbstractLevel} option that has no effect on {@link ManyLevelGuest}.
    */
@@ -144,6 +148,11 @@ declare interface GuestRef {
   unref: () => void
 }
 
+declare type GuestGetSync = (
+  key: Buffer,
+  options: AbstractGetOptions<Buffer, Buffer>
+) => Buffer | Uint8Array | undefined
+
 /**
  * Options for the {@link ManyLevelHost} constructor.
  */
@@ -158,17 +167,17 @@ declare interface HostOptions {
   /**
    * A function to be called before `db.put()` operations.
    */
-  preput?: (key: Buffer, value: Buffer, callback: NodeCallback<void>) => void
+  preput?: (key: Buffer, value: Buffer) => Promise<void>
 
   /**
    * A function to be called before `db.del()` operations.
    */
-  predel?: (key: Buffer, callback: NodeCallback<void>) => void
+  predel?: (key: Buffer) => Promise<void>
 
   /**
    * A function to be called before `db.batch()` operations.
    */
-  prebatch?: (operations: HostBatchOperation[], callback: NodeCallback<void>) => void
+  prebatch?: (operations: HostBatchOperation[]) => Promise<void>
 }
 
 declare type HostBatchOperation = HostBatchPutOperation | HostBatchDelOperation