Temporarily force exit + better constants + initialization cleanup

Author: andrewosh

index.js

@@ -10,6 +10,7 @@ const grpc = require('@grpc/grpc-js')
 const corestore = require('corestore')
 const SwarmNetworker = require('corestore-swarm-networking')
 
+
 const { rpc, loadMetadata } = require('hyperdrive-daemon-client')
 const constants = require('hyperdrive-daemon-client/lib/constants')
 
@@ -34,12 +35,6 @@ class HyperdriveDaemon extends EventEmitter {
     this.storage = opts.storage || constants.storage
     this.port = opts.port || constants.port
 
-    this.db = level(`${this.storage}/db`, { valueEncoding: 'json' })
-    const dbs = {
-      fuse: sub(this.db, 'fuse', { valueEncoding: 'json' }),
-      drives: sub(this.db, 'drives', { valueEncoding: 'json' })
-    }
-
     const corestoreOpts = {
       storage: path => raf(`${this.storage}/cores/${path}`),
       sparse: true,
@@ -60,21 +55,24 @@ class HyperdriveDaemon extends EventEmitter {
     }
     this.networking = new SwarmNetworker(this.corestore, networkOpts)
 
-    this.drives = new DriveManager(this.corestore, this.networking, dbs.drives, this.opts)
-    this.fuse = hyperfuse ? new FuseManager(this.megastore, this.drives, dbs.fuse, this.opts) : null
+    // Set in ready.
+    this.db = null
+    this.drives = null
+    this.fuse = null
+
     // Set in start.
     this.server = null
+    this._isMain = !!opts.main
     this._cleanup = null
 
-    this.drives.on('error', err => this.emit('error', err))
-    this.fuse.on('error', err => this.emit('error', err))
-
     this._isClosed = false
-    this._isReady = false
+    this._readyPromise = false
 
     this.ready = () => {
-      if (this._isReady) return Promise.resolve()
-      return this._ready()
+      if (this._isClosed) return Promise.resolve()
+      if (this._readyPromise) return this._readyPromise
+      this._readyPromise = this._ready()
+      return this._readyPromise
     }
   }
 
@@ -87,14 +85,24 @@ class HyperdriveDaemon extends EventEmitter {
       process.once(event, this._cleanup)
     }
 
-    return Promise.all([
-      this.db.open(),
-      this.networking.listen(),
+    this.networking.listen()
+
+    this.db = level(`${this.storage}/db`, { valueEncoding: 'json' })
+    const dbs = {
+      fuse: sub(this.db, 'fuse', { valueEncoding: 'json' }),
+      drives: sub(this.db, 'drives', { valueEncoding: 'json' })
+    }
+    this.drives = new DriveManager(this.corestore, this.networking, dbs.drives, this.opts)
+    this.fuse = hyperfuse ? new FuseManager(this.drives, dbs.fuse, this.opts) : null
+    this.drives.on('error', err => this.emit('error', err))
+    this.fuse.on('error', err => this.emit('error', err))
+
+    await Promise.all([
       this.drives.ready(),
       this.fuse ? this.fuse.ready() : Promise.resolve()
-    ]).then(() => {
-      this._ready = true
-    })
+    ])
+
+    this._isReady = true
   }
 
   async _loadMetadata () {
@@ -118,10 +126,11 @@ class HyperdriveDaemon extends EventEmitter {
   createMainHandlers () {
     return {
       stop: async (call) => {
-        await this.close()
+        await this.stop()
         setTimeout(() => {
           console.error('Daemon is exiting.')
           this.server.forceShutdown()
+          if (this._isMain) process.exit(0)
         }, 250)
         return new rpc.main.messages.StopResponse()
       },
@@ -137,7 +146,6 @@ class HyperdriveDaemon extends EventEmitter {
 
     if (this.fuse && this.fuse.fuseConfigured) await this.fuse.unmount()
     if (this.networking) await this.networking.close()
-    if (this.server) this.server.forceShutdown()
     await this.db.close()
 
     for (const event of STOP_EVENTS) {
@@ -171,11 +179,6 @@ class HyperdriveDaemon extends EventEmitter {
         return resolve()
       })
     })
-
-
-    async function close () {
-      await this.close()
-    }
   }
 }
 
@@ -236,7 +239,7 @@ function wrap (metadata, methods, opts) {
 
 if (require.main === module) {
   const opts = extractArguments()
-  const daemon = new HyperdriveDaemon(opts)
+  const daemon = new HyperdriveDaemon({ ...opts, main: true })
   daemon.start()
 } else {
   module.exports = HyperdriveDaemon

lib/drives/index.js

@@ -46,11 +46,16 @@ class DriveManager extends EventEmitter {
     this._watchers = new Map()
     this._sessionCounter = 0
 
-    this._ready = new Promise(async resolve => {
-      await this._reseed()
-      return resolve()
-    })
-    this.ready = () => this._ready
+    this._readyPromise = null
+
+    this.ready = () => {
+      if (this._readyPromise) return this._readyPromise
+      this._readyPromise = new Promise(async resolve => {
+        await this._reseed()
+        return resolve()
+      })
+      return this._readyPromise
+    }
   }
 
   async _reseed () {
@@ -306,19 +311,16 @@ class DriveManager extends EventEmitter {
         if (length !== 0) streamOpts.length = length
         streamOpts.start = start
 
-        console.log('STREAM OPTS:', streamOpts)
         const stream = drive.createReadStream(path, streamOpts)
 
         const rspMapper = map.obj(chunk => {
-          console.log('MAPPING RSP:', chunk)
           const rsp = new rpc.drive.messages.ReadStreamResponse()
           rsp.setChunk(chunk)
           return rsp
         })
 
         pump(stream, rspMapper, call, err => {
           if (err) log.error({ id, err }, 'createReadStream error')
-          console.log('AFTER PUMP HERE, err:', err)
         })
       },
 

lib/fuse/index.js

@@ -15,10 +15,9 @@ const constants = require('hyperdrive-daemon-client/lib/constants')
 const log = require('../log').child({ component: 'fuse-manager' })
 
 class FuseManager extends EventEmitter {
-  constructor (megastore, driveManager, db, opts) {
+  constructor (driveManager, db, opts) {
     super()
 
-    this.megastore = megastore
     this.driveManager = driveManager
     this.db = db
     this.opts = opts
@@ -27,7 +26,6 @@ class FuseManager extends EventEmitter {
     this._handlers = new Map()
 
     // Set in ready.
-
     this.fuseConfigured = false
     this._rootDrive = null
     this._rootMnt = null
@@ -444,8 +442,7 @@ class FuseManager extends EventEmitter {
 
         await this.unmount(mnt)
 
-        const rsp = rpc.fuse.messages.UnmountResponse()
-        return rsp
+        return new rpc.fuse.messages.UnmountResponse()
       },
 
       status: (call) => {

lib/log.js

@@ -3,13 +3,11 @@ const p = require('path')
 
 const pino = require('pino')
 const argv = require('yargs').argv
-
-// TODO: Move to a consts file.
-const LOGFILE = p.join(os.homedir(), '.hyperdrive', 'log.json')
+const constants = require('hyperdrive-daemon-client/lib/constants')
 
 // Forever will redirect stdout to the correct log file.
 module.exports = pino({
   name: 'hyperdrive',
   level: argv['log-level'] || 'info',
   enabled: true,
-}, (process.env['NODE_ENV'] === 'test') ? pino.destination(2) : LOGFILE)
+}, (process.env['NODE_ENV'] === 'test') ? pino.destination(2) : constants.structuredLog)

test/hyperdrive.js

@@ -49,7 +49,7 @@ test('can write/read a large file from a remote hyperdrive', async t => {
   t.end()
 })
 
-test.only('can write/read a file from a remote hyperdrive using stream methods', async t => {
+test('can write/read a file from a remote hyperdrive using stream methods', async t => {
   const { client, cleanup } = await createOne()
 
   try {
@@ -311,6 +311,13 @@ test.skip('watch cleans up after unexpected close', async t => {
   t.end()
 })
 
+// TODO: Figure out why the grpc server is not terminating.
+test.onFinish(() => {
+  setTimeout(() => {
+    process.exit(0)
+  }, 100)
+})
+
 function delay (ms) {
   return new Promise(resolve => setTimeout(resolve, ms))
 }