FIX: dyno crash on duplicate token insert + hardening from review

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",

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;
       }
     }
 

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"));

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");
+    });
   });
 });

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");

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);

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;
         }
       }