diff --git a/package-lock.json b/package-lock.json index 24eeca9..89aa9ed 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "groundcontrol", - "version": "3.2.0", + "version": "3.3.0", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "groundcontrol", - "version": "3.2.0", + "version": "3.3.0", "dependencies": { "body-parser": "^1.20.5", "cors": "^2.8.5", diff --git a/package.json b/package.json index fb75ce0..f2e86f4 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "groundcontrol", - "version": "3.2.0", + "version": "3.3.0", "description": "GroundControl push server API", "devDependencies": { "@types/node": "^22.12.0", diff --git a/src/controller/GroundController.ts b/src/controller/GroundController.ts index 85d0677..84e04e9 100644 --- a/src/controller/GroundController.ts +++ b/src/controller/GroundController.ts @@ -116,7 +116,9 @@ export class GroundController { token: body.token, os: body.os, }); - } catch (_) {} + } catch (error) { + if (error?.code !== "ER_DUP_ENTRY") throw error; // already subscribed + } } // todo: refactor into single batch save @@ -129,7 +131,9 @@ export class GroundController { token: body.token, os: body.os, }); - } catch (_) {} + } catch (error) { + if (error?.code !== "ER_DUP_ENTRY") throw error; // already subscribed + } } // todo: refactor into single batch save @@ -142,7 +146,9 @@ export class GroundController { token: body.token, os: body.os, }); - } catch (_) {} + } catch (error) { + if (error?.code !== "ER_DUP_ENTRY") throw error; // already subscribed + } } response.status(201).send(""); } @@ -167,24 +173,18 @@ export class GroundController { } for (const address of body.addresses) { - try { - const addressRecord = await this.tokenToAddressRepository.findOneBy({ os: body.os, token: body.token, address }); - await this.tokenToAddressRepository.remove(addressRecord); - } catch (_) {} + const addressRecord = await this.tokenToAddressRepository.findOneBy({ os: body.os, token: body.token, address }); + if (addressRecord) await this.tokenToAddressRepository.remove(addressRecord); } for (const hash of body.hashes) { - try { - const hashRecord = await this.tokenToHashRepository.findOneBy({ os: body.os, token: body.token, hash }); - await this.tokenToHashRepository.remove(hashRecord); - } catch (_) {} + const hashRecord = await this.tokenToHashRepository.findOneBy({ os: body.os, token: body.token, hash }); + if (hashRecord) await this.tokenToHashRepository.remove(hashRecord); } for (const txid of body.txids) { - try { - const txidRecord = await this.tokenToTxidRepository.findOneBy({ os: body.os, token: body.token, txid }); - await this.tokenToTxidRepository.remove(txidRecord); - } catch (_) {} + const txidRecord = await this.tokenToTxidRepository.findOneBy({ os: body.os, token: body.token, txid }); + if (txidRecord) await this.tokenToTxidRepository.remove(txidRecord); } response.status(201).send(""); @@ -260,23 +260,27 @@ export class GroundController { tokenConfig = new TokenConfiguration(); tokenConfig.token = body.token; tokenConfig.os = body.os; - } else { - if (typeof body.level_all !== "undefined") tokenConfig.level_all = !!body.level_all; - if (typeof body.level_transactions !== "undefined") tokenConfig.level_transactions = !!body.level_transactions; - if (typeof body.level_price !== "undefined") tokenConfig.level_price = !!body.level_price; - if (typeof body.level_news !== "undefined") tokenConfig.level_news = !!body.level_news; - if (typeof body.level_tips !== "undefined") tokenConfig.level_tips = !!body.level_tips; - if (typeof body.lang !== "undefined") tokenConfig.lang = String(body.lang); - if (typeof body.app_version !== "undefined") tokenConfig.app_version = String(body.app_version); - if (typeof body.redacted !== "undefined") tokenConfig.redacted = !!body.redacted; - tokenConfig.last_online = new Date(); + try { + await this.tokenConfigurationRepository.save(tokenConfig); + } catch (error) { + if (error?.code !== "ER_DUP_ENTRY") throw error; + // lost a create race: a concurrent request already inserted this (token, os); use that row + tokenConfig = await this.tokenConfigurationRepository.findOneBy({ token: body.token, os: body.os }); + if (!tokenConfig) throw error; + } } - try { - await this.tokenConfigurationRepository.save(tokenConfig); - } catch (error) { - console.warn(error.message); - } + if (typeof body.level_all !== "undefined") tokenConfig.level_all = !!body.level_all; + if (typeof body.level_transactions !== "undefined") tokenConfig.level_transactions = !!body.level_transactions; + if (typeof body.level_price !== "undefined") tokenConfig.level_price = !!body.level_price; + if (typeof body.level_news !== "undefined") tokenConfig.level_news = !!body.level_news; + if (typeof body.level_tips !== "undefined") tokenConfig.level_tips = !!body.level_tips; + if (typeof body.lang !== "undefined") tokenConfig.lang = String(body.lang); + if (typeof body.app_version !== "undefined") tokenConfig.app_version = String(body.app_version); + if (typeof body.redacted !== "undefined") tokenConfig.redacted = !!body.redacted; + tokenConfig.last_online = new Date(); + + await this.tokenConfigurationRepository.save(tokenConfig); response.status(200).send(""); } @@ -303,6 +307,7 @@ export class GroundController { if (error?.code !== "ER_DUP_ENTRY") throw error; // lost a create race: a concurrent request already inserted this (token, os); use that row tokenConfig = await this.tokenConfigurationRepository.findOneBy({ token: body.token, os: body.os }); + if (!tokenConfig) throw error; } } diff --git a/src/index.ts b/src/index.ts index 7c002b4..7c927b7 100644 --- a/src/index.ts +++ b/src/index.ts @@ -120,6 +120,18 @@ dataSource app.use(helmet.hidePoweredBy()); app.use(helmet.hsts()); + // rate limiter must be registered before the routes, otherwise it never runs: + // express dispatches in registration order and route handlers dont call next() + app.set("trust proxy", 1); + const rateLimit = require("express-rate-limit"); + const limiter = rateLimit({ + windowMs: 15 * 60 * 1000, + // sized for carrier-grade NAT: thousands of mobile users can share one egress IP, + // and this limiter only became effective once registered before the routes + max: 2000, + }); + app.use(limiter); + // register express routes from defined application routes Routes.forEach((route) => { (app as any)[route.method](route.route, (req: Request, res: Response, next: Function) => { @@ -140,14 +152,6 @@ dataSource }); }); - app.set("trust proxy", 1); - const rateLimit = require("express-rate-limit"); - const limiter = rateLimit({ - windowMs: 15 * 60 * 1000, - max: 100, - }); - app.use(limiter); - app.listen(process.env.PORT || 3001); console.log(require("fs").readFileSync("./bowie.txt").toString("ascii")); diff --git a/src/tests/GroundController.test.ts b/src/tests/GroundController.test.ts index 096b353..4a55690 100644 --- a/src/tests/GroundController.test.ts +++ b/src/tests/GroundController.test.ts @@ -261,6 +261,25 @@ describe("GroundController", () => { expect(mockRepository.save).not.toHaveBeenCalled(); expect(mockResponse.status).toHaveBeenCalledWith(201); }); + + it("should treat duplicate subscriptions as success", async () => { + const duplicateError: any = new Error("Duplicate entry"); + duplicateError.code = "ER_DUP_ENTRY"; + mockRepository.save.mockRejectedValue(duplicateError); + + await groundController.majorTomToGroundControl(mockRequest, mockResponse, mockNext); + + expect(mockRepository.save).toHaveBeenCalledTimes(3); + expect(mockResponse.status).toHaveBeenCalledWith(201); + }); + + it("should propagate non-duplicate save errors", async () => { + const dbError: any = new Error("Connection lost"); + dbError.code = "PROTOCOL_CONNECTION_LOST"; + mockRepository.save.mockRejectedValueOnce(dbError); + + await expect(groundController.majorTomToGroundControl(mockRequest, mockResponse, mockNext)).rejects.toThrow("Connection lost"); + }); }); describe("unsubscribe", () => { @@ -302,14 +321,13 @@ describe("GroundController", () => { expect(mockResponse.send).toHaveBeenCalledWith("token not provided"); }); - it("should handle records not found gracefully", async () => { + it("should skip removal when records are not found", async () => { mockRepository.findOneBy.mockResolvedValue(null); await groundController.unsubscribe(mockRequest, mockResponse, mockNext); expect(mockRepository.findOneBy).toHaveBeenCalledTimes(3); - expect(mockRepository.remove).toHaveBeenCalledTimes(3); - expect(mockRepository.remove).toHaveBeenCalledWith(null); + expect(mockRepository.remove).not.toHaveBeenCalled(); expect(mockResponse.status).toHaveBeenCalledWith(201); }); }); @@ -367,6 +385,50 @@ describe("GroundController", () => { expect(mockRepository.save).toHaveBeenCalled(); expect(mockResponse.status).toHaveBeenCalledWith(200); }); + + it("should apply settings when creating a new configuration", async () => { + mockRepository.findOneBy.mockResolvedValue(null); + + await groundController.setTokenConfiguration(mockRequest, mockResponse, mockNext); + + const lastSaved = mockRepository.save.mock.calls[mockRepository.save.mock.calls.length - 1][0]; + expect(lastSaved.level_all).toBe(false); + expect(lastSaved.level_price).toBe(false); + expect(lastSaved.lang).toBe("es"); + expect(lastSaved.app_version).toBe("2.0.0"); + expect(mockResponse.status).toHaveBeenCalledWith(200); + }); + + it("should apply settings to concurrently created configuration (ER_DUP_ENTRY)", async () => { + const concurrentlyCreatedConfig: any = { + token: "test-token", + os: "ios", + level_all: true, + lang: "en", + }; + const duplicateError: any = new Error("Duplicate entry"); + duplicateError.code = "ER_DUP_ENTRY"; + + mockRepository.findOneBy.mockResolvedValueOnce(null).mockResolvedValueOnce(concurrentlyCreatedConfig); + mockRepository.save.mockRejectedValueOnce(duplicateError).mockResolvedValue({}); + + await groundController.setTokenConfiguration(mockRequest, mockResponse, mockNext); + + expect(concurrentlyCreatedConfig.level_all).toBe(false); + expect(concurrentlyCreatedConfig.lang).toBe("es"); + expect(mockRepository.save).toHaveBeenLastCalledWith(concurrentlyCreatedConfig); + expect(mockResponse.status).toHaveBeenCalledWith(200); + }); + + it("should propagate non-duplicate save errors", async () => { + const dbError: any = new Error("Connection lost"); + dbError.code = "PROTOCOL_CONNECTION_LOST"; + + mockRepository.findOneBy.mockResolvedValue(null); + mockRepository.save.mockRejectedValueOnce(dbError); + + await expect(groundController.setTokenConfiguration(mockRequest, mockResponse, mockNext)).rejects.toThrow("Connection lost"); + }); }); describe("enqueue", () => { @@ -482,5 +544,15 @@ describe("GroundController", () => { await expect(groundController.getTokenConfiguration(mockRequest, mockResponse, mockNext)).rejects.toThrow("Connection lost"); }); + + it("should rethrow duplicate error if re-fetch finds nothing", async () => { + const duplicateError: any = new Error("Duplicate entry"); + duplicateError.code = "ER_DUP_ENTRY"; + + mockRepository.findOneBy.mockResolvedValue(null); + mockRepository.save.mockRejectedValueOnce(duplicateError); + + await expect(groundController.getTokenConfiguration(mockRequest, mockResponse, mockNext)).rejects.toThrow("Duplicate entry"); + }); }); }); diff --git a/src/worker-blockprocessor.ts b/src/worker-blockprocessor.ts index 5663cb8..896a822 100644 --- a/src/worker-blockprocessor.ts +++ b/src/worker-blockprocessor.ts @@ -145,12 +145,11 @@ dataSource try { await processBlock(nextBlockToProcess, sendQueueRepository); } catch (error) { - console.warn("exception when processing block:", error, "continuing as usuall"); - if (error.message.includes("socket hang up")) { - // issue fetching block from bitcoind - console.warn("retrying block number", nextBlockToProcess); - continue; // skip overwriting `LAST_PROCESSED_BLOCK` in `KeyValue` table - } + // dont advance LAST_PROCESSED_BLOCK: advancing would silently drop every notification + // in this block. transient errors (bitcoind hiccup, db down) resolve on retry + console.warn("exception when processing block:", error, "retrying block", nextBlockToProcess); + await new Promise((resolve) => setTimeout(resolve, 10_000, false)); + continue; } const end = +new Date(); console.log("took", (end - start) / 1000, "sec"); diff --git a/src/worker-processmempool.ts b/src/worker-processmempool.ts index 4312154..75adac2 100644 --- a/src/worker-processmempool.ts +++ b/src/worker-processmempool.ts @@ -7,16 +7,18 @@ import { components } from "./openapi/api"; import { LruCache } from "./lru-cache"; require("dotenv").config(); const url = require("url"); -let jayson = require("jayson/promise"); -let rpc = url.parse(process.env.BITCOIN_RPC); -let client = jayson.client.http(rpc); -const processedTxids = new LruCache(250000); if (!process.env.BITCOIN_RPC) { console.error("not all env variables set"); process.exit(); } +let jayson = require("jayson/promise"); +let rpc = url.parse(process.env.BITCOIN_RPC); +let client = jayson.client.http(rpc); + +const processedTxids = new LruCache(250000); + process .on("unhandledRejection", (reason, p) => { console.error(reason, "Unhandled Rejection at Promise", p); diff --git a/src/worker-sender.ts b/src/worker-sender.ts index b597b97..2f9980c 100644 --- a/src/worker-sender.ts +++ b/src/worker-sender.ts @@ -65,13 +65,16 @@ dataSource continue; } + // validate before querying: findOneBy with undefined values throws, and the poison record + // would crash-loop the worker since it never gets removed from the queue + if (!payload?.os || !payload?.token) { + process.env.VERBOSE && console.warn("no os or token in payload:", payload); + await sendQueueRepository.remove(record); + continue; + } + let tokenConfig = await tokenConfigurationRepository.findOneBy({ os: payload.os, token: payload.token }); if (!tokenConfig) { - if (!payload.os || !payload.token) { - process.env.VERBOSE && console.warn("no os or token in payload:", payload); - await sendQueueRepository.remove(record); - continue; - } tokenConfig = new TokenConfiguration(); tokenConfig.os = payload.os; tokenConfig.token = payload.token; @@ -81,6 +84,7 @@ dataSource if (error?.code !== "ER_DUP_ENTRY") throw error; // lost a create race with the API or another worker; use the existing row tokenConfig = await tokenConfigurationRepository.findOneBy({ os: payload.os, token: payload.token }); + if (!tokenConfig) throw error; } }