Add heart beat and retry closed process

packages/neuron-wallet/src/block-sync-renderer/task.ts

@@ -61,7 +61,7 @@ export const listener = async ({ type, id, channel, message }: WorkerMessage) =>
     }
 
     case 'unmount': {
-      if (!syncQueue) { return }
+      if (!syncQueue) { process.exit(0); return }
       logger.debug("Sync:\tstopping")
       await syncQueue.stopAndWait()
       syncQueue = null

packages/neuron-wallet/src/controllers/app/index.ts

@@ -15,6 +15,7 @@ import SyncApiController from 'controllers/sync-api'
 import { SETTINGS_WINDOW_TITLE } from 'utils/const'
 import IndexerService from 'services/indexer'
 import { stopCkbNode } from 'services/ckb-runner'
+import startMonitor from 'services/monitor'
 
 const app = electronApp
 
@@ -50,6 +51,8 @@ export default class AppController {
     SyncApiController.getInstance().mount()
 
     await this.openWindow()
+
+    startMonitor()
   }
 
   /**

packages/neuron-wallet/src/services/monitor/base.ts

@@ -0,0 +1,39 @@
+import logger from 'utils/logger'
+import { interval, timer, Subscription, race, from } from 'rxjs'
+import { map } from 'rxjs/operators'
+
+export default abstract class Monitor {
+  interval: Subscription | null = null
+
+  isReStarting: boolean = false
+
+  name: string = ''
+
+  abstract isLiving(): Promise<boolean>
+
+  abstract restart(): Promise<void>
+
+  startMonitor(intervalTime: number = 10000) {
+    this.interval = interval(intervalTime).subscribe(async () => {
+      if (this.isReStarting) {
+        return
+      }
+      const timeout = timer(intervalTime / 2).pipe(map(() => true))
+      const isLiving = await race(timeout, from(this.isLiving())).toPromise()
+      if (!isLiving) {
+        logger.info(`Monitor: is restarting ${this.name} process`)
+        this.isReStarting = true
+        try {
+          await this.restart()
+          logger.info(`Monitor: Restarting ${this.name} process success`)
+        } finally {
+          this.isReStarting = false
+        }
+      }
+    })
+  }
+
+  clearMonitor() {
+    this.interval?.unsubscribe()
+  }
+}

packages/neuron-wallet/src/services/monitor/ckb-indexer-monitor.ts

@@ -0,0 +1,24 @@
+import { resetSyncTask } from 'block-sync-renderer/index'
+import IndexerService from 'services/indexer'
+import { rpcRequest } from 'utils/rpc-request'
+import BaseMonitor from './base'
+
+export default class CkbIndexerMonitor extends BaseMonitor {
+  async isLiving(): Promise<boolean> {
+    try {
+      await rpcRequest(IndexerService.LISTEN_URI, { method: 'get_tip' })
+      return true
+    } catch (error) {
+      if (error?.code === 'ECONNREFUSED') {
+        return false
+      }
+      return true
+    }
+  }
+
+  async restart(): Promise<void> {
+    await resetSyncTask()
+  }
+
+  name: string = 'ckb-indexer'
+}

packages/neuron-wallet/src/services/monitor/ckb-monitor.ts

@@ -0,0 +1,14 @@
+import NodeService from '../node'
+import BaseMonitor from './base'
+
+export default class CkbMonitor extends BaseMonitor {
+  async isLiving(): Promise<boolean> {
+    return !(await NodeService.getInstance().isDefaultCKBNeedRestart())
+  }
+
+  async restart(): Promise<void> {
+    return NodeService.getInstance().startNode()
+  }
+
+  name: string = 'ckb'
+}

packages/neuron-wallet/src/services/monitor/index.ts

@@ -0,0 +1,10 @@
+import Base from './base'
+import CkbIndexerMonitor from './ckb-indexer-monitor'
+import CkbMonitor from './ckb-monitor'
+
+export default function startMonitor() {
+  const monitors = [new CkbIndexerMonitor(), new CkbMonitor()]
+  monitors.forEach((v: Base) => {
+    v.startMonitor()
+  })
+}

packages/neuron-wallet/src/services/node.ts

@@ -132,21 +132,38 @@ class NodeService {
       )
   }
 
-  public async tryStartNodeOnDefaultURI(): Promise<boolean> {
+  public async tryStartNodeOnDefaultURI() {
+    const isDefaultCKBNeedStart = await this.isDefaultCKBNeedRestart()
+    if (isDefaultCKBNeedStart) {
+      logger.info('CKB:\texternal RPC on default uri not detected, starting bundled CKB node.')
+      const redistReady = await redistCheck()
+      await (redistReady ? this.startNode() : this.showGuideDialog())
+    } else {
+      logger.info('CKB:\texternal RPC on default uri detected, skip starting bundled CKB node.')
+    }
+  }
+
+  public async isDefaultCKBNeedRestart() {
     let network = NetworksService.getInstance().getCurrent()
     if (network.remote !== BUNDLED_CKB_URL) {
       return false
     }
     try {
       await new RpcService(network.remote).getChain()
-      logger.info('CKB:\texternal RPC on default uri detected, skip starting bundled CKB node.')
       return false
     } catch (err) {
-      logger.info('CKB:\texternal RPC on default uri not detected, starting bundled CKB node.')
-      const redistReady = await redistCheck()
-      this.startedBundledNode = await (redistReady ? this.startNode() : this.showGuideDialog())
+      return true
+    }
+  }
 
-      return this.startedBundledNode
+  public async startNode() {
+    try {
+      await startCkbNode()
+      this.startedBundledNode = true
+    } catch (error) {
+      this.startedBundledNode = false
+      logger.info('CKB:\tfail to start bundled CKB with error:')
+      logger.error(error)
     }
   }
 
@@ -170,16 +187,6 @@ class NodeService {
         return false
       })
   }
-
-  private startNode = () => {
-    return startCkbNode()
-      .then(() => true)
-      .catch(err => {
-        logger.info('CKB:\tfail to start bundled CKB with error:')
-        logger.error(err)
-        return false
-      })
-  }
 }
 
 export default NodeService

packages/neuron-wallet/src/utils/rpc-request.ts

@@ -1,14 +1,41 @@
 import axios from 'axios'
 
+export const rpcRequest = async (
+  url: string,
+  options: {
+    method: string
+    params?: any
+  }
+): Promise<any[]> => {
+  const res = await axios.post<{ id: number; error?: any; result: any }[]>(
+    url,
+    {
+      id: 0,
+      jsonrpc: '2.0',
+      method: options.method,
+      params: options.params
+    },
+    {
+      headers: {
+        'content-type': 'application/json'
+      }
+    }
+  )
+  if (res.status !== 200) {
+    throw new Error(`indexer request failed with HTTP code ${res.status}`)
+  }
+  return res.data
+}
+
 export const rpcBatchRequest = async (
-  ckbIndexerUrl: string,
+  url: string,
   options: {
     method: string
     params?: any
   }[]
 ): Promise<any[]> => {
   const res = await axios.post<{ id: number; error?: any; result: any }[]>(
-    ckbIndexerUrl,
+    url,
     options.map((v, idx) => ({
       id: idx,
       jsonrpc: '2.0',
@@ -28,5 +55,6 @@ export const rpcBatchRequest = async (
 }
 
 export default {
-  rpcBatchRequest
+  rpcBatchRequest,
+  rpcRequest
 }

packages/neuron-wallet/tests/services/monitor.test.ts

@@ -0,0 +1,123 @@
+import Monitor from '../../src/services/monitor/base'
+import CkbMonitor from '../../src/services/monitor/ckb-monitor'
+import CkbIndexerMonitor from '../../src/services/monitor/ckb-indexer-monitor'
+
+const isDefaultCKBNeedRestartMock = jest.fn().mockResolvedValue(true)
+const startNodeMock = jest.fn()
+
+jest.mock('../../src/services/node', () => ({
+  getInstance() {
+    return {
+      isDefaultCKBNeedRestart: isDefaultCKBNeedRestartMock,
+      startNode: startNodeMock
+    }
+  }
+}))
+
+describe('ckb monitor', () => {
+  const monitor = new CkbMonitor()
+  it('is living', async () => {
+    const isLiving = await monitor.isLiving()
+    expect(isLiving).toBeFalsy()
+  })
+  it('restart', async () => {
+    await monitor.restart()
+    expect(startNodeMock).toHaveBeenCalled()
+  })
+})
+
+const rpcRequestMock = jest.fn()
+jest.mock('../../src/utils/rpc-request', () => ({
+  rpcRequest: () => rpcRequestMock()
+}))
+
+const resetSyncTaskMock = jest.fn()
+jest.mock('../../src/block-sync-renderer/index', () => ({
+  resetSyncTask: () => resetSyncTaskMock()
+}))
+
+describe('ckb indexer monitor', () => {
+  const monitor = new CkbIndexerMonitor()
+  describe('is living', () => {
+    it('rpc success', async () => {
+      const isLiving = await monitor.isLiving()
+      expect(isLiving).toBeTruthy()
+    })
+    it('rpc failed with ECONNREFUSED', async () => {
+      rpcRequestMock.mockRejectedValueOnce({ code: 'ECONNREFUSED' })
+      const isLiving = await monitor.isLiving()
+      expect(isLiving).toBeFalsy()
+    })
+    it('rpc failed not ECONNREFUSED', async () => {
+      rpcRequestMock.mockRejectedValueOnce({})
+      const isLiving = await monitor.isLiving()
+      expect(isLiving).toBeTruthy()
+    })
+  })
+  it('restart', async () => {
+    await monitor.restart()
+    expect(resetSyncTaskMock).toHaveBeenCalled()
+  })
+})
+
+const isLivingMock = jest.fn()
+const restartMock = jest.fn()
+class MonitorTest extends Monitor {
+  isLiving(): Promise<boolean> {
+    return isLivingMock()
+  }
+
+  restart(): Promise<void> {
+    return restartMock()
+  }
+}
+
+function wait(times: number) {
+  return new Promise((resolve) => {
+    setTimeout(() => { resolve(times) }, times)
+  })
+}
+describe('base monitor', () => {
+  const monitor = new MonitorTest()
+
+  beforeEach(() => {
+    isLivingMock.mockReset()
+    restartMock.mockReset()
+  })
+
+  describe('start monitor', () => {
+    it('is living', async () => {
+      isLivingMock.mockResolvedValue(true)
+      monitor.startMonitor(100)
+      await wait(200)
+      expect(isLivingMock).toHaveBeenCalled()
+      expect(restartMock).toHaveBeenCalledTimes(0)
+      monitor.clearMonitor()
+    })
+    it('not living', async () => {
+      isLivingMock.mockResolvedValue(true).mockResolvedValueOnce(false)
+      monitor.startMonitor(100)
+      await wait(200)
+      expect(isLivingMock).toHaveBeenCalled()
+      expect(restartMock).toHaveBeenCalled()
+      monitor.clearMonitor()
+    })
+    it('isLiving timeout', async () => {
+      isLivingMock.mockImplementation(() => wait(200))
+      monitor.startMonitor(100)
+      await wait(200)
+      expect(isLivingMock).toHaveBeenCalled()
+      expect(restartMock).toHaveBeenCalledTimes(0)
+      monitor.clearMonitor()
+    })
+    it('not living wait restart', async () => {
+      isLivingMock.mockResolvedValueOnce(false)
+      restartMock.mockImplementation(() => wait(400))
+      monitor.startMonitor(100)
+      await wait(400)
+      expect(isLivingMock).toHaveBeenCalled()
+      expect(restartMock).toHaveBeenCalledTimes(1)
+      monitor.clearMonitor()
+    })
+  })
+})