feat: add failover (#29)

* feat: add failover

* refactor: rename variable

* chore: add npmignore and increment version

Author: Cafe137

.npmignore

@@ -0,0 +1,6 @@
+src
+.github
+packaging
+.dockerignore
+Dockerfile
+tsconfig.json

README.md

@@ -27,10 +27,10 @@ curl http://localhost:9000 -H "Content-Type: application/json" -X POST --data '{
 When the text `Cache hit` appears, it indicates that a request was saved and immediately returned:
 
 ```txt
-[~] Key: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
-[~] Cache hit: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
-[~] Cache hit: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
-[~] Cache hit: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
+Key: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
+Cache hit: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
+Cache hit: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
+Cache hit: {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]}
 ```
 
 ## Benefits

package-lock.json

@@ -1,6 +1,6 @@
 {
     "name": "etherproxy",
-    "version": "1.0.0",
+    "version": "1.4.0",
     "description": "JSON-RPC reverse proxy tool designed for caching requests",
     "main": "dist/index.js",
     "bin": {
@@ -15,7 +15,7 @@
     "author": "@Cafe137",
     "license": "MIT",
     "dependencies": {
-        "cafe-utility": "^10.8.1",
+        "cafe-utility": "^11.0.2",
         "node-fetch": "^2.6.9"
     },
     "devDependencies": {

package.json

@@ -3,70 +3,110 @@
 import { Arrays, Objects } from 'cafe-utility'
 import { IncomingMessage, ServerResponse, createServer } from 'http'
 import fetch from 'node-fetch'
+import { Target, getHealthyTarget } from './target'
 import { RequestContext, ResponseContext } from './types'
 import { fetchWithTimeout, respondWithFetchPromise } from './utility'
 
+const PORT_ENV = 'ETHERPROXY_PORT'
+const EXPIRY_ENV = 'ETHERPROXY_EXPIRY'
+const TARGET_ENV = 'ETHERPROXY_TARGET'
+const DEFAULT_PORT = 9000
+const DEFAULT_EXPIRY = 2000
+const DEFAULT_TARGET = 'http://localhost:8545'
+
 main()
 
 function main() {
-    const port: number = parseInt(Arrays.getArgument(process.argv, 'port') as string) || parseInt((process.env.ETHERPROXY_PORT as string)) || 9000
-    const target: string = Arrays.getArgument(process.argv, 'target') as string || process.env.ETHERPROXY_TARGET as string || "http://localhost:8545"
-    const expiry: number = parseInt(Arrays.getArgument(process.argv, 'expiry') as string) || parseInt((process.env.ETHERPROXY_EXPIRY as string)) || 2000
-
+    const port = Arrays.getNumberArgument(process.argv, 'port', process.env, PORT_ENV) || DEFAULT_PORT
+    const target = Arrays.getArgument(process.argv, 'target', process.env, TARGET_ENV) || DEFAULT_TARGET
+    const expiry = Arrays.getNumberArgument(process.argv, 'expiry', process.env, EXPIRY_ENV) || DEFAULT_EXPIRY
     const fastIndex = Objects.createFastIndex()
+    const targets: Target[] = target.split(',').map(x => ({
+        url: x,
+        lastErrorAt: 0
+    }))
     const server = createServer(async (request: IncomingMessage, response: ServerResponse) => {
+        request.on('error', error => {
+            console.error(error)
+        })
+        response.on('error', error => {
+            console.error(error)
+        })
         if (request.url === '/health' || request.url === '/readiness') {
-            try {
-                await fetch(target, { timeout: 10_000 })
-                response.statusCode = 200
-                response.end('200 OK')
-            } catch (error) {
-                console.error(error)
-                response.statusCode = 503
-                response.end('503 Service Unavailable')
+            for (let i = 0; i < targets.length; i++) {
+                const target = getHealthyTarget(targets)
+                try {
+                    await fetch(target.url, { timeout: 10_000 })
+                    response.statusCode = 200
+                    response.end('200 OK')
+                    return
+                } catch (error) {
+                    target.lastErrorAt = Date.now()
+                    console.error(error)
+                }
             }
+            response.statusCode = 503
+            response.end('503 Service Unavailable')
             return
         }
         const chunks: Buffer[] = []
         request.on('data', (chunk: Buffer) => {
             chunks.push(chunk)
         })
         request.on('end', async () => {
-            try {
-                const context: RequestContext = {
-                    method: request.method || 'GET',
-                    url: target,
-                    headers: request.headers as Record<string, string>,
-                    body: Buffer.concat(chunks).toString('utf-8')
-                }
-                const parsedBody = JSON.parse(context.body)
-                const id = parsedBody.id
-                delete parsedBody.id
-                const key = JSON.stringify(parsedBody)
-                const cachedPromise = Objects.getFromFastIndexWithExpiracy(fastIndex, key) as Promise<ResponseContext>
-                if (cachedPromise) {
-                    process.stdout.write(`[~] Cache hit: ${key}\n`)
-                    await respondWithFetchPromise(id, response, cachedPromise)
-                    return
+            for (let i = 0; i < targets.length; i++) {
+                const target = getHealthyTarget(targets)
+                try {
+                    const context: RequestContext = {
+                        method: request.method || 'GET',
+                        url: target.url,
+                        headers: request.headers as Record<string, string>,
+                        body: Buffer.concat(chunks).toString('utf-8')
+                    }
+                    const parsedBody = JSON.parse(context.body)
+                    const id = parsedBody.id
+                    delete parsedBody.id
+                    const key = `${target.url}_${JSON.stringify(parsedBody)}`
+                    const cachedPromise = Objects.getFromFastIndexWithExpiracy(
+                        fastIndex,
+                        key
+                    ) as Promise<ResponseContext>
+                    if (cachedPromise) {
+                        process.stdout.write(`Cache hit: ${key}\n`)
+                        const successful = await respondWithFetchPromise(id, response, cachedPromise)
+                        if (successful) {
+                            return
+                        } else {
+                            target.lastErrorAt = Date.now()
+                            continue
+                        }
+                    }
+                    process.stdout.write(`Key: ${key}\n`)
+                    delete context.headers.host
+                    delete context.headers['user-agent']
+                    delete context.headers['content-length']
+                    const responsePromise = fetchWithTimeout(context.url, {
+                        method: context.method,
+                        headers: context.headers,
+                        body: context.body
+                    })
+                    Objects.pushToFastIndexWithExpiracy(fastIndex as any, key, responsePromise, expiry)
+                    const successful = await respondWithFetchPromise(id, response, responsePromise)
+                    if (successful) {
+                        return
+                    } else {
+                        target.lastErrorAt = Date.now()
+                        continue
+                    }
+                } catch (error) {
+                    target.lastErrorAt = Date.now()
+                    console.error(error)
                 }
-                process.stdout.write(`[~] Key: ${key}\n`)
-                delete context.headers.host
-                delete context.headers['user-agent']
-                delete context.headers['content-length']
-                const responsePromise = fetchWithTimeout(context.url, {
-                    method: context.method,
-                    headers: context.headers,
-                    body: context.body
-                })
-                Objects.pushToFastIndexWithExpiracy(fastIndex as any, key, responsePromise, expiry)
-                await respondWithFetchPromise(id, response, responsePromise)
-            } catch (error) {
-                console.error(error)
-                response.statusCode = 503
-                response.end('503 Service Unavailable')
             }
+            response.statusCode = 503
+            response.end('503 Service Unavailable')
         })
     })
     server.listen(port)
-    console.log(`[~] Etherproxy is running on port ${port}`)
+    console.log(`Etherproxy is running on port ${port}`)
 }

src/index.ts

@@ -0,0 +1,12 @@
+import { Dates } from 'cafe-utility'
+
+export type Target = {
+    url: string
+    lastErrorAt: number
+}
+
+export function getHealthyTarget(targets: Target[]): Target {
+    const healthyIfLastErrorIsBefore = Date.now() - Dates.hours(2)
+    const healthyTargets = targets.filter(x => x.lastErrorAt < healthyIfLastErrorIsBefore)
+    return healthyTargets[0] || targets[0]
+}

src/target.ts

@@ -6,12 +6,10 @@ export async function respondWithFetchPromise(
     id: number,
     response: ServerResponse,
     promise: Promise<ResponseContext | null>
-) {
+): Promise<boolean> {
     const context = await promise
     if (!context) {
-        response.statusCode = 503
-        response.end('503 Service Unavailable')
-        return
+        return false
     }
     for (const [key, value] of Object.entries(context.headers)) {
         const lowerCaseKey = key.toLowerCase()
@@ -23,6 +21,7 @@ export async function respondWithFetchPromise(
     response.statusCode = context.status
     context.json.id = id
     response.end(JSON.stringify(context.json))
+    return true
 }
 
 export async function fetchWithTimeout(url: string, options: RequestInit): Promise<ResponseContext | null> {