Merge pull request #136 from interledgerjs/ko-server-improvements

feat: server reliability improvements

Author: kincaidoneil

package-lock.json

@@ -70,7 +70,7 @@
     "ilp-plugin-mini-accounts": "^4.2.0",
     "mocha": "^5.1.1",
     "mocha-typescript": "^1.1.14",
-    "nyc": "^13.1.0",
+    "nyc": "^15.0.0",
     "puppeteer": "^1.19.0",
     "sinon": "^6.0.1",
     "source-map-support": "^0.5.6",

package.json

@@ -1,5 +1,4 @@
 import { EventEmitter } from 'events'
-import * as assert from 'assert'
 import createLogger from 'ilp-logger'
 import { DataAndMoneyStream } from './stream'
 import * as IlpPacket from 'ilp-packet'
@@ -29,15 +28,12 @@ import { Reader } from 'oer-utils'
 import { CongestionController } from './util/congestion'
 import { Plugin } from './util/plugin-interface'
 import {
-  LongValue,
-  longFromValue,
   maxLong,
   minLong,
   minLongs,
   countDigits,
   checkedAdd,
   checkedSubtract,
-  checkedMultiply,
   multiplyDivideFloor
 } from './util/long'
 import * as Long from 'long'
@@ -55,6 +51,8 @@ const DEFAULT_MINIMUM_EXCHANGE_RATE_PRECISION = 3
 const TEST_PACKET_MAX_ATTEMPTS = 20
 
 export interface ConnectionOpts {
+  /** Token in the ILP address uniquely identifying this connection */
+  connectionId?: string,
   /** Ledger plugin (V2) */
   plugin: Plugin,
   /** ILP Address of the remote entity */
@@ -93,6 +91,17 @@ export interface ConnectionOpts {
    * When omitted, use a timeout of 30 seconds.
    */
   getExpiry?: (destination: string) => Date,
+  /**
+   * Callback for the consumer to perform accounting and choose to fulfill an incoming ILP Prepare,
+   * given the amount received, a unique identifier for the packet, and `connectionTag`.
+   *
+   * If the returned Promise is resolved, the ILP Prepare will be fulfilled; if it is rejected,
+   * the ILP Prepare will be rejected. The ILP Fulfill will be immediately sent back after
+   * the Promise is resolved.
+   *
+   * Note: a misbehaving sender can trigger duplicate packetIds, which should be ignored and rejected.
+   */
+  shouldFulfill?: (packetAmount: Long, packetId: Buffer, connectionTag?: string) => Promise<void>,
 }
 
 export interface BuildConnectionOpts extends ConnectionOpts {
@@ -181,15 +190,21 @@ export class Connection extends EventEmitter {
 
   // TODO use longs for byte offsets
   protected remoteMaxOffset: number
+  protected _incomingHold: Long
   protected _totalReceived: Long
   protected _totalSent: Long
   protected _totalDelivered: Long
   protected _lastPacketExchangeRate: Rational
   protected getExpiry: (destination: string) => Date
+  protected shouldFulfill?: (packetAmount: Long, packetId: Buffer, connectionTag?: string) => Promise<void>
 
   constructor (opts: NewConnectionOpts) {
     super()
-    this.connectionId = uuid()
+
+    // Use the same connectionId for logging on both client & server
+    const lastAddressSegment = opts.destinationAccount ? opts.destinationAccount.split('.').slice(-1)[0] : undefined
+    this.connectionId = (opts.connectionId || lastAddressSegment || uuid()).replace(/[-_]/g, '').slice(0, 8)
+
     this.plugin = opts.plugin
     this._sourceAccount = opts.sourceAccount
     this._sourceAssetCode = opts.assetCode
@@ -213,6 +228,7 @@ export class Connection extends EventEmitter {
       ? undefined
       : Rational.fromNumber(opts.exchangeRate, true)
     this.getExpiry = opts.getExpiry || defaultGetExpiry
+    this.shouldFulfill = opts.shouldFulfill
     this.idleTimeout = opts.idleTimeout || DEFAULT_IDLE_TIMEOUT
     this.lastActive = new Date()
 
@@ -239,6 +255,7 @@ export class Connection extends EventEmitter {
 
     this.remoteMaxOffset = this.maxBufferedData
 
+    this._incomingHold = Long.UZERO
     this._totalReceived = Long.UZERO
     this._totalSent = Long.UZERO
     this._totalDelivered = Long.UZERO
@@ -476,7 +493,8 @@ export class Connection extends EventEmitter {
 
   /**
    * (Internal) Handle incoming ILP Prepare packets.
-   * This will automatically fulfill all valid and expected Prepare packets.
+   * This will automatically fulfill all valid and expected Prepare packets, or
+   * defer to custom application logic using the `shouldFulfill` callback, if provided.
    * It passes the incoming money and/or data to the relevant streams.
    * @private
    */
@@ -628,9 +646,16 @@ export class Connection extends EventEmitter {
       }
     }
 
-    // Add incoming amounts to each stream
-    for (let { stream, amount } of amountsToReceive) {
-      stream._addToIncoming(amount)
+    this.addIncomingHold(incomingAmount)
+
+    // Allow consumer to choose to fulfill each packet and/or perform other logic before fulfilling
+    if (this.shouldFulfill && incomingAmount.greaterThan(0)) {
+      const packetId = await cryptoHelper.generateIncomingPacketId(this.sharedSecret, requestPacket.sequence)
+      await this.shouldFulfill(incomingAmount, packetId, this.connectionTag).catch(async err => {
+        this.removeIncomingHold(incomingAmount)
+        this.log.debug('application declined to fulfill packet %s:', requestPacket.sequence, err)
+        await throwFinalApplicationError()
+      })
     }
 
     // Tell peer about closed streams and how much each stream can receive
@@ -655,12 +680,18 @@ export class Connection extends EventEmitter {
       }
     }
 
+    // Add incoming amounts to each stream
+    for (let { stream, amount } of amountsToReceive) {
+      stream._addToIncoming(amount)
+    }
+
     // TODO make sure the queued frames aren't too big
     responseFrames = responseFrames.concat(this.queuedFrames)
     this.queuedFrames = []
 
     // Return fulfillment and response packet
     const responsePacket = new Packet(requestPacket.sequence, IlpPacketType.Fulfill, incomingAmount, responseFrames)
+    this.removeIncomingHold(incomingAmount)
     this.addTotalReceived(incomingAmount)
     this.log.trace('fulfilling prepare with fulfillment: %h and response packet: %j', fulfillment, responsePacket)
     return {
@@ -1588,19 +1619,33 @@ export class Connection extends EventEmitter {
 
   private bumpIdle (): void { this.lastActive = new Date() }
 
-  private addTotalReceived (value: Long): void {
-    const result = checkedAdd(this._totalReceived, value)
+  private addIncomingHold (value: Long): void {
+    let result = checkedAdd(this._totalReceived, this._incomingHold)
+    result = checkedAdd(result.sum, value)
+
     if (result.overflow) {
       const err = new IlpPacket.Errors.BadRequestError('Total received exceeded MaxUint64')
       err['ilpErrorMessage'] = err.message
       /* tslint:disable-next-line:no-floating-promises */
       this.destroy(err)
       throw err
     } else {
-      this._totalReceived = result.sum
+      this._incomingHold = result.sum
     }
   }
 
+  private removeIncomingHold (value: Long): void {
+    // As long as this is called after `addIncomingHold` for the same amount,
+    // this should never underflow
+    this._incomingHold = checkedSubtract(this._incomingHold, value).difference
+  }
+
+  private addTotalReceived (value: Long): void {
+    // As long as this is called after `addIncomingHold` for the same amount,
+    // this should never overflow
+    this._totalReceived = checkedAdd(this._totalReceived, value).sum
+  }
+
   private addTotalSent (value: Long): void {
     const result = checkedAdd(this._totalSent, value)
     if (result.overflow) {

src/connection.ts

@@ -11,6 +11,7 @@ export {
 const TOKEN_LENGTH = 18
 const ENCRYPTION_KEY_STRING = Buffer.from('ilp_stream_encryption', 'utf8')
 const FULFILLMENT_GENERATION_STRING = Buffer.from('ilp_stream_fulfillment', 'utf8')
+const PACKET_ID_STRING = Buffer.from('ilp_stream_packet_id', 'utf8')
 export const ENCRYPTION_OVERHEAD = 28
 
 export function generateToken (): Buffer {
@@ -32,3 +33,7 @@ export function generateFulfillmentKey (sharedSecret: Buffer): Promise<Buffer> {
 export function generateFulfillment (fulfillmentKey: Buffer, data: Buffer): Promise<Buffer> {
   return hmac(fulfillmentKey, data)
 }
+
+export function generateIncomingPacketId (sharedSecret: Buffer, sequence: Long): Promise<Buffer> {
+  return hmac(sharedSecret, Buffer.concat([PACKET_ID_STRING, Buffer.from(sequence.toBytes())]))
+}

src/crypto.ts

@@ -1,13 +1,8 @@
-import { EventEmitter } from 'events'
 import * as ILDCP from 'ilp-protocol-ildcp'
 import * as IlpPacket from 'ilp-packet'
 import createLogger from 'ilp-logger'
 import './util/formatters'
-import * as cryptoHelper from './crypto'
 import { Connection, ConnectionOpts } from './connection'
-import { Plugin } from './util/plugin-interface'
-
-const CONNECTION_ID_REGEX = /^[a-zA-Z0-9~_-]+$/
 
 export { Connection } from './connection'
 export { DataAndMoneyStream } from './stream'

src/index.ts

@@ -59,7 +59,8 @@ export class ServerConnectionPool {
       const conn = await Connection.build({
         ...this.connectionOpts,
         sharedSecret,
-        connectionTag
+        connectionTag,
+        connectionId: id
       })
       log.debug('got incoming packet for new connection: %s%s', id, (connectionTag ? ' (connectionTag: ' + connectionTag + ')' : ''))
       try {

src/pool.ts

@@ -8,9 +8,16 @@ import { ServerConnectionPool } from './pool'
 import { Plugin } from './util/plugin-interface'
 
 const CONNECTION_ID_REGEX = /^[a-zA-Z0-9~_-]+$/
+const DEFAULT_DISCONNECT_DELAY = 100
 
 export interface ServerOpts extends ConnectionOpts {
   serverSecret?: Buffer
+
+  /**
+   * Number of milliseconds to wait between closing the server and disconnecting
+   * the plugin so packets may be safely returned
+   */
+  disconnectDelay?: number
 }
 
 /**
@@ -32,6 +39,8 @@ export class Server extends EventEmitter {
   protected enablePadding?: boolean
   protected connected: boolean
   protected connectionOpts: ConnectionOpts
+  protected pendingRequests: Promise<any> = Promise.resolve()
+  protected disconnectDelay: number
   private pool: ServerConnectionPool
 
   constructor (opts: ServerOpts) {
@@ -42,6 +51,7 @@ export class Server extends EventEmitter {
     this.connectionOpts = Object.assign({}, opts, {
       serverSecret: undefined
     }) as ConnectionOpts
+    this.disconnectDelay = opts.disconnectDelay || DEFAULT_DISCONNECT_DELAY
     this.connected = false
   }
 
@@ -64,7 +74,12 @@ export class Server extends EventEmitter {
     if (this.connected && this.plugin.isConnected()) {
       return
     }
-    this.plugin.registerDataHandler(this.handleData.bind(this))
+    this.plugin.registerDataHandler(data => {
+      this.emit('_incoming_prepare')
+      const request = this.handleData(data)
+      this.pendingRequests = this.pendingRequests.then(() => request.finally())
+      return request
+    })
     await this.plugin.connect()
     const { clientAddress, assetCode, assetScale } = await ILDCP.fetch(this.plugin.sendData.bind(this.plugin))
     this.serverAccount = clientAddress
@@ -87,9 +102,28 @@ export class Server extends EventEmitter {
    * End all connections and disconnect the plugin
    */
   async close (): Promise<void> {
+    // Stop handling new requests, and return T99 while the connection is closing.
+    // If an F02 unreachable was returned on new packets: clients would immediately destroy the connection
+    // If an F99 was returned on on new packets: clients would retry with no backoff
+    this.plugin.deregisterDataHandler()
+    this.plugin.registerDataHandler(async () => IlpPacket.serializeIlpReject({
+      code: IlpPacket.Errors.codes.T99_APPLICATION_ERROR,
+      triggeredBy: this.serverAccount,
+      message: 'Shutting down server',
+      data: Buffer.alloc(0)
+    }))
+
+    // Wait for in-progress requests to finish so all Fulfills are returned
+    await this.pendingRequests
+    // Allow the plugin time to send the reply packets back before disconnecting it
+    await new Promise(r => setTimeout(r, this.disconnectDelay))
+
+    // Gracefully close the connection and all streams
     await this.pool.close()
+
     this.plugin.deregisterDataHandler()
     await this.plugin.disconnect()
+
     this.emit('_close')
     this.connected = false
   }