Merge pull request #3389 from fosrl/dev

1.19.4-s.3

server/lib/dbRetry.ts

@@ -0,0 +1,83 @@
+import logger from "@server/logger";
+
+const MAX_RETRIES = 5;
+const BASE_DELAY_MS = 50;
+
+/**
+ * Detect transient errors that are safe to retry (connection drops, deadlocks,
+ * serialization failures). PostgreSQL deadlocks (40P01) are always safe to
+ * retry: the database guarantees exactly one winner per deadlock pair, so the
+ * loser just needs to try again.
+ */
+export function isTransientError(error: any): boolean {
+    if (!error) return false;
+
+    const message = (error.message || "").toLowerCase();
+    const causeMessage = (error.cause?.message || "").toLowerCase();
+    const code = error.code || error.cause?.code || "";
+
+    // Connection timeout / terminated
+    if (
+        message.includes("connection timeout") ||
+        message.includes("connection terminated") ||
+        message.includes("timeout exceeded when trying to connect") ||
+        causeMessage.includes("connection terminated unexpectedly") ||
+        causeMessage.includes("connection timeout")
+    ) {
+        return true;
+    }
+
+    // PostgreSQL deadlock detected - always safe to retry (one winner guaranteed)
+    if (code === "40P01" || message.includes("deadlock")) {
+        return true;
+    }
+
+    // PostgreSQL serialization failure
+    if (code === "40001") {
+        return true;
+    }
+
+    // ECONNRESET, ECONNREFUSED, EPIPE, ETIMEDOUT
+    if (
+        code === "ECONNRESET" ||
+        code === "ECONNREFUSED" ||
+        code === "EPIPE" ||
+        code === "ETIMEDOUT"
+    ) {
+        return true;
+    }
+
+    return false;
+}
+
+/**
+ * Simple retry wrapper with exponential backoff for transient errors
+ * (deadlocks, connection timeouts, unexpected disconnects).
+ */
+export async function withRetry<T>(
+    operation: () => Promise<T>,
+    context: string,
+    maxRetries: number = MAX_RETRIES,
+    baseDelayMs: number = BASE_DELAY_MS
+): Promise<T> {
+    let attempt = 0;
+    while (true) {
+        try {
+            return await operation();
+        } catch (error: any) {
+            if (isTransientError(error) && attempt < maxRetries) {
+                attempt++;
+                const baseDelay = Math.pow(2, attempt - 1) * baseDelayMs;
+                const jitter = Math.random() * baseDelay;
+                const delay = baseDelay + jitter;
+                logger.warn(
+                    `Transient DB error in ${context}, retrying attempt ${attempt}/${maxRetries} after ${delay.toFixed(0)}ms`,
+                    { code: error?.code ?? error?.cause?.code }
+                );
+                await new Promise((resolve) => setTimeout(resolve, delay));
+                continue;
+            }
+            throw error;
+        }
+    }
+}

server/lib/exitNodes/subnet.ts

@@ -1,10 +1,31 @@
-import { db, exitNodes } from "@server/db";
+import { db, exitNodes, Transaction } from "@server/db";
 import config from "@server/lib/config";
 import { findNextAvailableCidr } from "@server/lib/ip";
+import { lockManager } from "#dynamic/lib/lock";
 
-export async function getNextAvailableSubnet(): Promise<string> {
+/**
+ * Reserves the next available exit node subnet.
+ *
+ * Exit node subnets must never overlap with one another - regardless of
+ * which org(s) they belong to - since HA exit nodes can end up routing for
+ * the same org. This acquires a lock that the caller MUST release (via the
+ * returned `release`) only after the chosen address has been durably
+ * persisted (e.g. after the enclosing transaction commits), otherwise
+ * concurrent callers can race and pick the same subnet.
+ */
+export async function getNextAvailableSubnet(
+    trx: Transaction | typeof db = db
+): Promise<{ value: string; release: () => Promise<void> }> {
+    const lockKey = "exit-node-subnet-allocation";
+    const acquired = await lockManager.acquireLockWithRetry(lockKey, 6000);
+    if (!acquired) {
+        throw new Error(`Failed to acquire lock: ${lockKey}`);
+    }
+    const release = () => lockManager.releaseLock(lockKey, acquired);
+
+    try {
         // Get all existing subnets from routes table
-    const existingAddresses = await db
+        const existingAddresses = await trx
             .select({
                 address: exitNodes.address
             })
@@ -26,5 +47,9 @@ export async function getNextAvailableSubnet(): Promise<string> {
             ".1" +
             "/" +
             subnet.split("/")[1];
-    return subnet;
+        return { value: subnet, release };
+    } catch (e) {
+        await release();
+        throw e;
+    }
 }

server/lib/rebuildClientAssociations.ts

@@ -45,6 +45,7 @@ import {
 } from "@server/routers/client/targets";
 import { lockManager } from "#dynamic/lib/lock";
 import { rebuildQueue } from "#dynamic/lib/rebuildQueue";
+import { withRetry, isTransientError } from "@server/lib/dbRetry";
 import {
     checkOrgRebuildRateLimit,
     decrementOrgRebuildCount,
@@ -285,10 +286,20 @@ export async function rebuildClientAssociationsFromSiteResource(
 ) {
     await incrementOrgRebuildCount(siteResource.orgId);
     try {
-        return await lockManager.withLock(
+        // The whole locked rebuild is idempotent (it diffs full expected vs.
+        // actual state each time), so on a transient DB error it's safe to
+        // retry the entire thing rather than just the failed query.
+        return await withRetry(
+            () =>
+                lockManager.withLock(
                     `rebuild-client-associations:site-resource:${siteResource.siteResourceId}`,
-            () => rebuildClientAssociationsFromSiteResourceImpl(siteResource),
+                    () =>
+                        rebuildClientAssociationsFromSiteResourceImpl(
+                            siteResource
+                        ),
                     REBUILD_ASSOCIATIONS_LOCK_TTL_MS
+                ),
+            `rebuildClientAssociationsFromSiteResource:${siteResource.siteResourceId}`
         );
     } catch (err: any) {
         if (
@@ -304,6 +315,17 @@ export async function rebuildClientAssociationsFromSiteResource(
             });
             return { mergedAllClients: [] };
         }
+        if (isTransientError(err)) {
+            logger.warn(
+                `rebuildClientAssociations: transient DB error rebuilding site resource ${siteResource.siteResourceId} persisted after retries, queuing for deferred processing:`,
+                err
+            );
+            await rebuildQueue.enqueue({
+                type: "site-resource",
+                id: siteResource.siteResourceId
+            });
+            return { mergedAllClients: [] };
+        }
         throw err;
     } finally {
         await decrementOrgRebuildCount(siteResource.orgId);
@@ -463,6 +485,7 @@ async function rebuildClientAssociationsFromSiteResourceImpl(
         await trx
             .insert(clientSiteResourcesAssociationsCache)
             .values(clientSiteResourcesToInsert)
+            .onConflictDoNothing()
             .returning();
         logger.debug(
             `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteResourceId=${siteResource.siteResourceId} inserted clientSiteResource associations`
@@ -510,6 +533,7 @@ async function rebuildClientAssociationsFromSiteResourceImpl(
     for (const site of sitesToProcess) {
         const siteId = site.siteId;
 
+        try {
             logger.debug(
                 `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] processing siteId=${siteId} for siteResourceId=${siteResource.siteResourceId}`
             );
@@ -539,11 +563,14 @@ async function rebuildClientAssociationsFromSiteResourceImpl(
                               subnet: clients.subnet
                           })
                           .from(clients)
-                      .where(inArray(clients.clientId, existingClientSiteIds))
+                          .where(
+                              inArray(clients.clientId, existingClientSiteIds)
+                          )
                     : [];
 
             const otherResourceClientIds =
-            clientsFromOtherResourcesBySite.get(siteId) ?? new Set<number>();
+                clientsFromOtherResourcesBySite.get(siteId) ??
+                new Set<number>();
 
             logger.debug(
                 `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} otherResourceClientIds=[${[...otherResourceClientIds].join(", ")}] mergedAllClientIds=[${mergedAllClientIds.join(", ")}]`
@@ -577,6 +604,7 @@ async function rebuildClientAssociationsFromSiteResourceImpl(
                 await trx
                     .insert(clientSitesAssociationsCache)
                     .values(clientSitesToInsert)
+                    .onConflictDoNothing()
                     .returning();
                 logger.debug(
                     `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserted clientSite associations`
@@ -625,6 +653,21 @@ async function rebuildClientAssociationsFromSiteResourceImpl(
                 clientSitesToRemove,
                 trx
             );
+        } catch (err) {
+            // Don't let a failure on one site abort processing of every
+            // other site queued after it in this run. Since we're not
+            // re-throwing, the outer wrapper's retry/requeue logic never
+            // sees this failure, so explicitly queue this resource for a
+            // follow-up pass to reconcile whatever this site didn't get to.
+            logger.error(
+                `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} failed while processing site for siteResourceId=${siteResource.siteResourceId}, continuing with remaining sites and queuing a follow-up pass:`,
+                err
+            );
+            await rebuildQueue.enqueue({
+                type: "site-resource",
+                id: siteResource.siteResourceId
+            });
+        }
     }
 
     // Handle subnet proxy target updates for the resource associations
@@ -917,7 +960,7 @@ export async function updateClientSiteDestinations(
 
     for (const site of sitesData) {
         if (!site.sites.subnet) {
-            logger.warn(`Site ${site.sites.siteId} has no subnet, skipping`);
+            logger.debug(`Site ${site.sites.siteId} has no subnet, skipping`);
             continue;
         }
 
@@ -1656,10 +1699,17 @@ export async function rebuildClientAssociationsFromClient(
     await incrementOrgRebuildCount(client.orgId);
     try {
         const trx = primaryDb;
-        return await lockManager.withLock(
+        // The whole locked rebuild is idempotent (it diffs full expected vs.
+        // actual state each time), so on a transient DB error it's safe to
+        // retry the entire thing rather than just the failed query.
+        return await withRetry(
+            () =>
+                lockManager.withLock(
                     `rebuild-client-associations:client:${client.clientId}`,
                     () => rebuildClientAssociationsFromClientImpl(client, trx),
                     REBUILD_ASSOCIATIONS_LOCK_TTL_MS
+                ),
+            `rebuildClientAssociationsFromClient:${client.clientId}`
         );
     } catch (err: any) {
         if (
@@ -1675,6 +1725,17 @@ export async function rebuildClientAssociationsFromClient(
             });
             return;
         }
+        if (isTransientError(err)) {
+            logger.warn(
+                `rebuildClientAssociations: transient DB error rebuilding client ${client.clientId} persisted after retries, queuing for deferred processing:`,
+                err
+            );
+            await rebuildQueue.enqueue({
+                type: "client",
+                id: client.clientId
+            });
+            return;
+        }
         throw err;
     } finally {
         await decrementOrgRebuildCount(client.orgId);
@@ -1826,12 +1887,15 @@ async function rebuildClientAssociationsFromClientImpl(
 
     // Insert new associations
     if (resourcesToAdd.length > 0) {
-        await trx.insert(clientSiteResourcesAssociationsCache).values(
+        await trx
+            .insert(clientSiteResourcesAssociationsCache)
+            .values(
                 resourcesToAdd.map((siteResourceId) => ({
                     clientId: client.clientId,
                     siteResourceId
                 }))
-        );
+            )
+            .onConflictDoNothing();
     }
 
     // Remove old associations
@@ -1869,12 +1933,15 @@ async function rebuildClientAssociationsFromClientImpl(
 
     // Insert new site associations
     if (sitesToAdd.length > 0) {
-        await trx.insert(clientSitesAssociationsCache).values(
+        await trx
+            .insert(clientSitesAssociationsCache)
+            .values(
                 sitesToAdd.map((siteId) => ({
                     clientId: client.clientId,
                     siteId
                 }))
-        );
+            )
+            .onConflictDoNothing();
     }
 
     // Remove old site associations

server/lib/rebuildQueue.ts

@@ -1,10 +1,14 @@
 import logger from "@server/logger";
+import { isTransientError } from "@server/lib/dbRetry";
 
 export type RebuildJobType = "site-resource" | "client";
 
 export interface RebuildJob {
     type: RebuildJobType;
     id: number;
+    // Number of times this job has already been re-queued after a transient
+    // failure. Absent/0 means it has not failed yet.
+    attempt?: number;
 }
 
 export interface RebuildJobHandlers {
@@ -24,6 +28,10 @@ export interface RebuildQueueManager {
 // retried shortly after against fresh DB state.
 const POLL_INTERVAL_MS = 500;
 const BATCH_SIZE = 5;
+// A job that fails with a transient DB error gets re-queued with backoff
+// instead of being dropped, up to this many times.
+const MAX_JOB_ATTEMPTS = 5;
+const JOB_RETRY_BASE_DELAY_MS = 1000;
 
 function dedupeKey(job: RebuildJob): string {
     return `${job.type}:${job.id}`;
@@ -106,12 +114,31 @@ class InMemoryRebuildQueue implements RebuildQueueManager {
                         `Rebuild queue: completed ${job.type}:${job.id}`
                     );
                 } catch (err) {
+                    const attempt = (job.attempt ?? 0) + 1;
+                    if (isTransientError(err) && attempt <= MAX_JOB_ATTEMPTS) {
+                        const delay =
+                            JOB_RETRY_BASE_DELAY_MS * Math.pow(2, attempt - 1);
+                        logger.warn(
+                            `Rebuild queue: job ${job.type}:${job.id} hit a transient error (attempt ${attempt}/${MAX_JOB_ATTEMPTS}), re-queuing in ${delay}ms:`,
+                            err
+                        );
+                        setTimeout(() => {
+                            this.enqueue({ ...job, attempt }).catch(
+                                (enqueueErr) =>
+                                    logger.error(
+                                        `Rebuild queue: failed to re-queue ${job.type}:${job.id} after transient error:`,
+                                        enqueueErr
+                                    )
+                            );
+                        }, delay);
+                    } else {
                         logger.error(
                             `Rebuild queue: job ${job.type}:${job.id} threw an error:`,
                             err
                         );
                     }
                 }
+            }
         } finally {
             this.processing = false;
         }

server/private/lib/rebuildQueue.ts

@@ -14,12 +14,16 @@
 import { redis } from "#private/lib/redis";
 import { lockManager } from "#private/lib/lock";
 import logger from "@server/logger";
+import { isTransientError } from "@server/lib/dbRetry";
 
 export type RebuildJobType = "site-resource" | "client";
 
 export interface RebuildJob {
     type: RebuildJobType;
     id: number;
+    // Number of times this job has already been re-queued after a transient
+    // failure. Absent/0 means it has not failed yet.
+    attempt?: number;
 }
 
 export interface RebuildJobHandlers {
@@ -43,6 +47,11 @@ const PROCESSOR_LOCK_TTL_MS = 120000 * BATCH_SIZE + 30000; // ~630 s
 
 const POLL_INTERVAL_MS = 500;
 
+// A job that fails with a transient DB error gets re-queued with backoff
+// instead of being dropped, up to this many times.
+const MAX_JOB_ATTEMPTS = 5;
+const JOB_RETRY_BASE_DELAY_MS = 1000;
+
 class RedisRebuildQueue {
     private processingStarted = false;
 
@@ -180,12 +189,35 @@ class RedisRebuildQueue {
                                 `Rebuild queue: completed ${job.type}:${job.id}`
                             );
                         } catch (err) {
+                            const attempt = (job.attempt ?? 0) + 1;
+                            if (
+                                isTransientError(err) &&
+                                attempt <= MAX_JOB_ATTEMPTS
+                            ) {
+                                const delay =
+                                    JOB_RETRY_BASE_DELAY_MS *
+                                    Math.pow(2, attempt - 1);
+                                logger.warn(
+                                    `Rebuild queue: job ${job.type}:${job.id} hit a transient error (attempt ${attempt}/${MAX_JOB_ATTEMPTS}), re-queuing in ${delay}ms:`,
+                                    err
+                                );
+                                setTimeout(() => {
+                                    this.enqueue({ ...job, attempt }).catch(
+                                        (enqueueErr) =>
+                                            logger.error(
+                                                `Rebuild queue: failed to re-queue ${job.type}:${job.id} after transient error:`,
+                                                enqueueErr
+                                            )
+                                    );
+                                }, delay);
+                            } else {
                                 logger.error(
                                     `Rebuild queue: job ${job.type}:${job.id} threw an error:`,
                                     err
                                 );
                             }
                         }
+                    }
                 },
                 PROCESSOR_LOCK_TTL_MS
             );

server/private/lib/redis.ts

@@ -894,6 +894,19 @@ class RegionalRedisManager {
         return opts;
     }
 
+    // The regional Redis StatefulSet's "redis" service pins to pod redis-0
+    // (primary). The replica (redis-1) is only reachable through the
+    // per-pod headless service: <svc>.<namespace>.svc.cluster.local ->
+    // redis-1.redis-headless.<namespace>.svc.cluster.local. Returns null
+    // if the configured host doesn't match that pattern (e.g. local dev),
+    // in which case callers should fall back to the primary for reads.
+    private getReplicaHost(primaryHost: string): string | null {
+        const match = primaryHost.match(/^redis\.([^.]+)\.svc\.cluster\.local$/);
+        if (!match) return null;
+        const namespace = match[1];
+        return `redis-1.redis-headless.${namespace}.svc.cluster.local`;
+    }
+
     private initializeClients(): void {
         const cfg = this.getConfig();
         const baseOpts = {
@@ -907,35 +920,42 @@ class RegionalRedisManager {
 
         try {
             this.writeClient = new Redis(baseOpts);
-            // redis-1 (replica) handles reads; fall back to primary if not resolvable
-            this.readClient = new Redis({
-                ...baseOpts,
-                host: cfg.host!.replace(/^(.*?)(\.\S+)$/, (_, h, rest) => {
-                    // Derive replica hostname from the headless service pattern:
-                    // redis.redis.svc.cluster.local -> redis-1.redis-headless.redis.svc.cluster.local
-                    // If it doesn't look like a k8s service, just use the same host
-                    return h + rest;
-                })
-            });
 
-            // For simplicity use same host for both; callers can always read from primary
+            const replicaHost = this.getReplicaHost(cfg.host!);
-            // The real replica routing is handled by the StatefulSet headless service
+            this.readClient = replicaHost
-            this.readClient = this.writeClient;
+                ? new Redis({ ...baseOpts, host: replicaHost })
+                : this.writeClient;
 
             this.writeClient.on("ready", () => {
-                logger.info("Regional Redis client ready");
+                logger.info("Regional Redis write client ready");
                 this.isHealthy = true;
             });
             this.writeClient.on("error", (err) => {
-                logger.error("Regional Redis client error:", err);
+                logger.error("Regional Redis write client error:", err);
                 this.isHealthy = false;
             });
             this.writeClient.on("reconnecting", () => {
-                logger.info("Regional Redis client reconnecting...");
+                logger.info("Regional Redis write client reconnecting...");
                 this.isHealthy = false;
             });
 
-            logger.info("Regional Redis client initialized");
+            if (this.readClient !== this.writeClient) {
+                this.readClient.on("ready", () => {
+                    logger.info("Regional Redis read client ready");
+                });
+                this.readClient.on("error", (err) => {
+                    logger.error("Regional Redis read client error:", err);
+                });
+                this.readClient.on("reconnecting", () => {
+                    logger.info("Regional Redis read client reconnecting...");
+                });
+            }
+
+            logger.info(
+                replicaHost
+                    ? `Regional Redis client initialized (reads routed to replica ${replicaHost})`
+                    : "Regional Redis client initialized (no replica resolvable, reads routed to primary)"
+            );
         } catch (error) {
             logger.error("Failed to initialize regional Redis client:", error);
             this.isEnabled = false;
@@ -1041,11 +1061,14 @@ class RegionalRedisManager {
 
     public async disconnect(): Promise<void> {
         try {
+            if (this.readClient && this.readClient !== this.writeClient) {
+                await this.readClient.quit();
+            }
+            this.readClient = null;
             if (this.writeClient) {
                 await this.writeClient.quit();
                 this.writeClient = null;
             }
-            this.readClient = null;
             logger.info("Regional Redis client disconnected");
         } catch (error) {
             logger.error("Error disconnecting regional Redis client:", error);

server/private/routers/gerbil/createExitNode.ts

@@ -29,7 +29,8 @@ export async function createExitNode(
         .where(eq(exitNodes.publicKey, publicKey));
     let exitNode: ExitNode;
     if (!exitNodeQuery) {
-        const address = await getNextAvailableSubnet();
+        const { value: address, release } = await getNextAvailableSubnet();
+        try {
             // TODO: eventually we will want to get the next available port so that we can multiple exit nodes
             // const listenPort = await getNextAvailablePort();
             const listenPort = config.getRawConfig().gerbil.start_port;
@@ -60,6 +61,9 @@ export async function createExitNode(
             logger.info(
                 `Created new exit node ${exitNode.name} with address ${exitNode.address} and port ${exitNode.listenPort}`
             );
+        } finally {
+            await release();
+        }
     } else {
         // update the reachable at
         [exitNode] = await db

server/private/routers/remoteExitNode/createRemoteExitNode.ts

@@ -114,8 +114,6 @@ export async function createRemoteExitNode(
         }
 
         const secretHash = await hashPassword(secret);
-        // const address = await getNextAvailableSubnet();
-        const address = "100.89.140.1/24"; // FOR NOW LETS HARDCODE THESE ADDRESSES
 
         const [existingRemoteExitNode] = await db
             .select()
@@ -191,13 +189,27 @@ export async function createRemoteExitNode(
             );
         }
 
+        // If this remote exit node isn't already backing an exit node in
+        // another org, we're about to create a brand new one. Reserve a
+        // subnet for it up front so the allocation lock is held across the
+        // whole insert - this guarantees exit node subnets never overlap,
+        // even under concurrent creation, which matters for HA setups.
+        let releaseSubnetLock: (() => Promise<void>) | null = null;
+        let newExitNodeAddress: string | null = null;
+        if (!existingExitNode) {
+            const { value, release } = await getNextAvailableSubnet();
+            newExitNodeAddress = value;
+            releaseSubnetLock = release;
+        }
+
+        try {
             await db.transaction(async (trx) => {
                 if (!existingExitNode) {
                     const [res] = await trx
                         .insert(exitNodes)
                         .values({
                             name: remoteExitNodeId,
-                        address,
+                            address: newExitNodeAddress!,
                             endpoint: "",
                             publicKey: "",
                             listenPort: 0,
@@ -274,6 +286,9 @@ export async function createRemoteExitNode(
                     }
                 }
             });
+        } finally {
+            await releaseSubnetLock?.();
+        }
 
         const token = generateSessionToken();
         await createRemoteExitNodeSession(token, remoteExitNodeId);

server/routers/gerbil/createExitNode.ts

@@ -13,7 +13,8 @@ export async function createExitNode(
     const [exitNodeQuery] = await db.select().from(exitNodes).limit(1);
     let exitNode: ExitNode;
     if (!exitNodeQuery) {
-        const address = await getNextAvailableSubnet();
+        const { value: address, release } = await getNextAvailableSubnet();
+        try {
             // TODO: eventually we will want to get the next available port so that we can multiple exit nodes
             // const listenPort = await getNextAvailablePort();
             const listenPort = config.getRawConfig().gerbil.start_port;
@@ -44,6 +45,9 @@ export async function createExitNode(
             logger.info(
                 `Created new exit node ${exitNode.name} with address ${exitNode.address} and port ${exitNode.listenPort}`
             );
+        } finally {
+            await release();
+        }
     } else {
         // update the existing exit node
         [exitNode] = await db

server/routers/newt/buildConfiguration.ts

@@ -52,13 +52,13 @@ export async function buildClientConfigurationForNewtClient(
             clientsRes
                 .filter((client) => {
                     if (!client.clients.pubKey) {
-                        logger.warn(
+                        logger.debug(
                             `Client ${client.clients.clientId} has no public key, skipping`
                         );
                         return false;
                     }
                     if (!client.clients.subnet) {
-                        logger.warn(
+                        logger.debug(
                             `Client ${client.clients.clientId} has no subnet, skipping`
                         );
                         return false;

server/routers/newt/handleNewtDisconnectingMessage.ts

@@ -43,6 +43,6 @@ export const handleNewtDisconnectingMessage: MessageHandler = async (
             );
         });
     } catch (error) {
-        logger.error("Error handling disconnecting message", { error });
+        logger.error("Error handling site disconnecting message", error);
     }
 };

server/routers/newt/handleNewtPingMessage.ts

@@ -5,8 +5,21 @@ import { Newt } from "@server/db";
 import { eq } from "drizzle-orm";
 import logger from "@server/logger";
 import { sendNewtSyncMessage } from "./sync";
+import semver from "semver";
 import { recordSitePing } from "./pingAccumulator";
 
+const NEWT_SUPPORTS_SYNC_VERSION = ">=1.14.0";
+const PONG = {
+    message: {
+        type: "pong",
+        data: {
+            timestamp: new Date().toISOString()
+        }
+    },
+    broadcast: false,
+    excludeSender: false
+};
+
 /**
  * Handles ping messages from newt clients.
  *
@@ -37,6 +50,14 @@ export const handleNewtPingMessage: MessageHandler = async (context) => {
     // cross-region latency to the database.
     recordSitePing(newt.siteId);
 
+    if (
+        newt.version &&
+        !semver.satisfies(newt.version, NEWT_SUPPORTS_SYNC_VERSION)
+    ) {
+        // Newt does not support the sync message so not checking - stop here -
+        return PONG;
+    }
+
     // Check config version and sync if stale.
     const configVersion = await getClientConfigVersion(newt.newtId);
 
@@ -65,14 +86,5 @@ export const handleNewtPingMessage: MessageHandler = async (context) => {
         await sendNewtSyncMessage(newt, site);
     }
 
-    return {
+    return PONG;
-        message: {
-            type: "pong",
-            data: {
-                timestamp: new Date().toISOString()
-            }
-        },
-        broadcast: false,
-        excludeSender: false
-    };
 };

server/routers/newt/pingAccumulator.ts

@@ -3,6 +3,7 @@ import { sites, clients, olms } from "@server/db";
 import { and, eq, inArray } from "drizzle-orm";
 import logger from "@server/logger";
 import { fireSiteOnlineAlert } from "@server/lib/alerts";
+import { withRetry } from "@server/lib/dbRetry";
 
 /**
  * Ping Accumulator
@@ -22,8 +23,6 @@ import { fireSiteOnlineAlert } from "@server/lib/alerts";
  */
 
 const FLUSH_INTERVAL_MS = 10_000; // Flush every 10 seconds
-const MAX_RETRIES = 5;
-const BASE_DELAY_MS = 50;
 
 // ── Site (newt) pings ──────────────────────────────────────────────────
 // Map of siteId -> latest ping timestamp (unix seconds)
@@ -266,85 +265,7 @@ export async function flushPingsToDb(): Promise<void> {
 }
 
 // ── Retry / Error Helpers ──────────────────────────────────────────────
-
+// See @server/lib/dbRetry for the shared withRetry/isTransientError helpers.
-/**
- * Simple retry wrapper with exponential backoff for transient errors
- * (deadlocks, connection timeouts, unexpected disconnects).
- *
- * PostgreSQL deadlocks (40P01) are always safe to retry: the database
- * guarantees exactly one winner per deadlock pair, so the loser just needs
- * to try again. MAX_RETRIES is intentionally higher than typical connection
- * retry budgets to give deadlock victims enough chances to succeed.
- */
-async function withRetry<T>(
-    operation: () => Promise<T>,
-    context: string
-): Promise<T> {
-    let attempt = 0;
-    while (true) {
-        try {
-            return await operation();
-        } catch (error: any) {
-            if (isTransientError(error) && attempt < MAX_RETRIES) {
-                attempt++;
-                const baseDelay = Math.pow(2, attempt - 1) * BASE_DELAY_MS;
-                const jitter = Math.random() * baseDelay;
-                const delay = baseDelay + jitter;
-                logger.warn(
-                    `Transient DB error in ${context}, retrying attempt ${attempt}/${MAX_RETRIES} after ${delay.toFixed(0)}ms`,
-                    { code: error?.code ?? error?.cause?.code }
-                );
-                await new Promise((resolve) => setTimeout(resolve, delay));
-                continue;
-            }
-            throw error;
-        }
-    }
-}
-
-/**
- * Detect transient errors that are safe to retry.
- */
-function isTransientError(error: any): boolean {
-    if (!error) return false;
-
-    const message = (error.message || "").toLowerCase();
-    const causeMessage = (error.cause?.message || "").toLowerCase();
-    const code = error.code || error.cause?.code || "";
-
-    // Connection timeout / terminated
-    if (
-        message.includes("connection timeout") ||
-        message.includes("connection terminated") ||
-        message.includes("timeout exceeded when trying to connect") ||
-        causeMessage.includes("connection terminated unexpectedly") ||
-        causeMessage.includes("connection timeout")
-    ) {
-        return true;
-    }
-
-    // PostgreSQL deadlock detected - always safe to retry (one winner guaranteed)
-    if (code === "40P01" || message.includes("deadlock")) {
-        return true;
-    }
-
-    // PostgreSQL serialization failure
-    if (code === "40001") {
-        return true;
-    }
-
-    // ECONNRESET, ECONNREFUSED, EPIPE, ETIMEDOUT
-    if (
-        code === "ECONNRESET" ||
-        code === "ECONNREFUSED" ||
-        code === "EPIPE" ||
-        code === "ETIMEDOUT"
-    ) {
-        return true;
-    }
-
-    return false;
-}
 
 // ── Lifecycle ──────────────────────────────────────────────────────────
 

server/routers/newt/sync.ts

@@ -9,45 +9,45 @@ import {
 import { canCompress } from "@server/lib/clientVersionChecks";
 
 export async function sendNewtSyncMessage(newt: Newt, site: Site) {
-    // const {
+    const {
-    //     tcpTargets,
+        tcpTargets,
-    //     udpTargets,
+        udpTargets,
-    //     validHealthCheckTargets,
+        validHealthCheckTargets,
-    //     browserGatewayTargets,
+        browserGatewayTargets,
-    //     remoteExitNodeSubnets
+        remoteExitNodeSubnets
-    // } = await buildTargetConfigurationForNewtClient(site.siteId);
+    } = await buildTargetConfigurationForNewtClient(site.siteId);
-    // let exitNode: ExitNode | undefined;
+    let exitNode: ExitNode | undefined;
-    // if (site.exitNodeId) {
+    if (site.exitNodeId) {
-    //     [exitNode] = await db
+        [exitNode] = await db
-    //         .select()
+            .select()
-    //         .from(exitNodes)
+            .from(exitNodes)
-    //         .where(eq(exitNodes.exitNodeId, site.exitNodeId))
+            .where(eq(exitNodes.exitNodeId, site.exitNodeId))
-    //         .limit(1);
+            .limit(1);
-    // }
+    }
-    // const { peers, targets } = await buildClientConfigurationForNewtClient(
+    const { peers, targets } = await buildClientConfigurationForNewtClient(
-    //     site,
+        site,
-    //     exitNode
+        exitNode
-    // );
+    );
-    // await sendToClient(
+    await sendToClient(
-    //     newt.newtId,
+        newt.newtId,
-    //     {
+        {
-    //         type: "newt/sync",
+            type: "newt/sync",
-    //         data: {
+            data: {
-    //             proxyTargets: {
+                proxyTargets: {
-    //                 udp: udpTargets,
+                    udp: udpTargets,
-    //                 tcp: tcpTargets
+                    tcp: tcpTargets
-    //             },
+                },
-    //             healthCheckTargets: validHealthCheckTargets,
+                healthCheckTargets: validHealthCheckTargets,
-    //             peers: peers,
+                peers: peers,
-    //             clientTargets: targets,
+                clientTargets: targets,
-    //             browserGatewayTargets: browserGatewayTargets,
+                browserGatewayTargets: browserGatewayTargets,
-    //             remoteExitNodeSubnets: remoteExitNodeSubnets
+                remoteExitNodeSubnets: remoteExitNodeSubnets
-    //         }
+            }
-    //     },
+        },
-    //     {
+        {
-    //         compress: canCompress(newt.version, "newt")
+            compress: canCompress(newt.version, "newt")
-    //     }
+        }
-    // ).catch((error) => {
+    ).catch((error) => {
-    //     logger.warn(`Error sending newt sync message:`, error);
+        logger.warn(`Error sending newt sync message:`, error);
-    // });
+    });
 }

server/routers/olm/buildConfiguration.ts

@@ -161,7 +161,7 @@ export async function buildSiteConfigurationForOlmClient(
         }
 
         if (!site.subnet) {
-            logger.warn(`Site ${site.siteId} has no subnet, skipping`);
+            logger.debug(`Site ${site.siteId} has no subnet, skipping`);
             continue;
         }
 

server/routers/olm/handleOlmDisconnectingMessage.ts

@@ -6,7 +6,9 @@ import logger from "@server/logger";
 /**
  * Handles disconnecting messages from clients to show disconnected in the ui
  */
-export const handleOlmDisconnectingMessage: MessageHandler = async (context) => {
+export const handleOlmDisconnectingMessage: MessageHandler = async (
+    context
+) => {
     const { message, client: c, sendToClient } = context;
     const olm = c as Olm;
 
@@ -29,6 +31,6 @@ export const handleOlmDisconnectingMessage: MessageHandler = async (context) =>
             })
             .where(eq(clients.clientId, olm.clientId));
     } catch (error) {
-        logger.error("Error handling disconnecting message", { error });
+        logger.error("Error handling client disconnecting message", error);
     }
 };

src/app/[orgId]/settings/(private)/billing/page.tsx

@@ -290,7 +290,13 @@ export default function BillingPage() {
                     setHasSubscription(
                         tierSub.subscription.status === "active"
                     );
-                    setIsTrial(tierSub.subscription.expiresAt != null);
+                    // expiresAt is only meaningful while the trial hasn't
+                    // actually run out yet; a stale row with a past
+                    // expiresAt should no longer be treated as a live trial
+                    const expiresAt = tierSub.subscription.expiresAt;
+                    setIsTrial(
+                        expiresAt != null && expiresAt * 1000 > Date.now()
+                    );
                 }
 
                 // Find license subscription
@@ -1057,12 +1063,13 @@ export default function BillingPage() {
                     </SettingsSectionDescription>
                 </SettingsSectionHeader>
                 <SettingsSectionBody>
-                    <div className="grid grid-cols-1 md:grid-cols-2 gap-6">
+                    <div className="grid grid-cols-1 md:grid-cols-4 gap-6">
                         {/* Current Usage */}
-                        <div className="border rounded-lg p-4">
+                        <div className="border rounded-lg p-4 md:col-span-1">
                             <div className="text-sm text-muted-foreground mb-2">
                                 {t("billingCurrentUsage") || "Current Usage"}
                             </div>
+                            <div className="flex flex-col items-start gap-1">
                                 <div className="flex items-baseline gap-2">
                                     <span className="text-3xl font-semibold">
                                         {getUserCount()}
@@ -1070,8 +1077,9 @@ export default function BillingPage() {
                                     <span className="text-lg">
                                         {t("billingUsers") || "Users"}
                                     </span>
+                                </div>
                                 {hasSubscription && getPricePerUser() > 0 && (
-                                    <div className="text-sm text-muted-foreground mt-1">
+                                    <div className="text-sm text-muted-foreground">
                                         x ${getPricePerUser()} / month = $
                                         {getUserCount() * getPricePerUser()} /
                                         month
@@ -1081,7 +1089,7 @@ export default function BillingPage() {
                         </div>
 
                         {/* Maximum Limits */}
-                        <div className="border rounded-lg p-4">
+                        <div className="border rounded-lg p-4 md:col-span-3">
                             <div className="text-sm text-muted-foreground mb-3">
                                 {t("billingMaximumLimits") || "Maximum Limits"}
                             </div>