From 5fc5a3ebca45f64648abf784238d7e256f94f491 Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 2 Jul 2026 11:49:49 -0400 Subject: [PATCH 1/8] Adjust spacing --- .../settings/(private)/billing/page.tsx | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/src/app/[orgId]/settings/(private)/billing/page.tsx b/src/app/[orgId]/settings/(private)/billing/page.tsx index b3d1bb351..aa121838f 100644 --- a/src/app/[orgId]/settings/(private)/billing/page.tsx +++ b/src/app/[orgId]/settings/(private)/billing/page.tsx @@ -1057,21 +1057,23 @@ export default function BillingPage() { -
+
{/* Current Usage */} -
+
{t("billingCurrentUsage") || "Current Usage"}
-
- - {getUserCount()} - - - {t("billingUsers") || "Users"} - +
+
+ + {getUserCount()} + + + {t("billingUsers") || "Users"} + +
{hasSubscription && getPricePerUser() > 0 && ( -
+
x ${getPricePerUser()} / month = $ {getUserCount() * getPricePerUser()} / month @@ -1081,7 +1083,7 @@ export default function BillingPage() {
{/* Maximum Limits */} -
+
{t("billingMaximumLimits") || "Maximum Limits"}
From 1bf3d2cdd67e219e976514d1da1f801bd16d4b57 Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 2 Jul 2026 12:10:20 -0400 Subject: [PATCH 2/8] Add back the sync with semver --- server/routers/newt/handleNewtPingMessage.ts | 32 +++++--- server/routers/newt/sync.ts | 82 ++++++++++---------- 2 files changed, 63 insertions(+), 51 deletions(-) diff --git a/server/routers/newt/handleNewtPingMessage.ts b/server/routers/newt/handleNewtPingMessage.ts index 86489bd67..13052dcc1 100644 --- a/server/routers/newt/handleNewtPingMessage.ts +++ b/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 { - message: { - type: "pong", - data: { - timestamp: new Date().toISOString() - } - }, - broadcast: false, - excludeSender: false - }; + return PONG; }; diff --git a/server/routers/newt/sync.ts b/server/routers/newt/sync.ts index 6c3ab0a0f..30e2d59f8 100644 --- a/server/routers/newt/sync.ts +++ b/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 { - // tcpTargets, - // udpTargets, - // validHealthCheckTargets, - // browserGatewayTargets, - // remoteExitNodeSubnets - // } = await buildTargetConfigurationForNewtClient(site.siteId); - // let exitNode: ExitNode | undefined; - // if (site.exitNodeId) { - // [exitNode] = await db - // .select() - // .from(exitNodes) - // .where(eq(exitNodes.exitNodeId, site.exitNodeId)) - // .limit(1); - // } - // const { peers, targets } = await buildClientConfigurationForNewtClient( - // site, - // exitNode - // ); - // await sendToClient( - // newt.newtId, - // { - // type: "newt/sync", - // data: { - // proxyTargets: { - // udp: udpTargets, - // tcp: tcpTargets - // }, - // healthCheckTargets: validHealthCheckTargets, - // peers: peers, - // clientTargets: targets, - // browserGatewayTargets: browserGatewayTargets, - // remoteExitNodeSubnets: remoteExitNodeSubnets - // } - // }, - // { - // compress: canCompress(newt.version, "newt") - // } - // ).catch((error) => { - // logger.warn(`Error sending newt sync message:`, error); - // }); + const { + tcpTargets, + udpTargets, + validHealthCheckTargets, + browserGatewayTargets, + remoteExitNodeSubnets + } = await buildTargetConfigurationForNewtClient(site.siteId); + let exitNode: ExitNode | undefined; + if (site.exitNodeId) { + [exitNode] = await db + .select() + .from(exitNodes) + .where(eq(exitNodes.exitNodeId, site.exitNodeId)) + .limit(1); + } + const { peers, targets } = await buildClientConfigurationForNewtClient( + site, + exitNode + ); + await sendToClient( + newt.newtId, + { + type: "newt/sync", + data: { + proxyTargets: { + udp: udpTargets, + tcp: tcpTargets + }, + healthCheckTargets: validHealthCheckTargets, + peers: peers, + clientTargets: targets, + browserGatewayTargets: browserGatewayTargets, + remoteExitNodeSubnets: remoteExitNodeSubnets + } + }, + { + compress: canCompress(newt.version, "newt") + } + ).catch((error) => { + logger.warn(`Error sending newt sync message:`, error); + }); } From f87e136f6b2e8be0d0e152e31e1ac78a39fcaa14 Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 2 Jul 2026 20:54:59 -0400 Subject: [PATCH 3/8] Unique subnets for exit nodes --- server/lib/exitNodes/subnet.ts | 73 ++++--- .../private/routers/gerbil/createExitNode.ts | 64 ++++--- .../remoteExitNode/createRemoteExitNode.ts | 181 ++++++++++-------- server/routers/gerbil/createExitNode.ts | 64 ++++--- 4 files changed, 215 insertions(+), 167 deletions(-) diff --git a/server/lib/exitNodes/subnet.ts b/server/lib/exitNodes/subnet.ts index 49e28bd57..8c4f3e99e 100644 --- a/server/lib/exitNodes/subnet.ts +++ b/server/lib/exitNodes/subnet.ts @@ -1,30 +1,55 @@ -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 { - // Get all existing subnets from routes table - const existingAddresses = await db - .select({ - address: exitNodes.address - }) - .from(exitNodes); - - const addresses = existingAddresses.map((a) => a.address); - let subnet = findNextAvailableCidr( - addresses, - config.getRawConfig().gerbil.block_size, - config.getRawConfig().gerbil.subnet_group - ); - if (!subnet) { - throw new Error("No available subnets remaining in space"); +/** + * 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 }> { + 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); - // replace the last octet with 1 - subnet = - subnet.split(".").slice(0, 3).join(".") + - ".1" + - "/" + - subnet.split("/")[1]; - return subnet; + try { + // Get all existing subnets from routes table + const existingAddresses = await trx + .select({ + address: exitNodes.address + }) + .from(exitNodes); + + const addresses = existingAddresses.map((a) => a.address); + let subnet = findNextAvailableCidr( + addresses, + config.getRawConfig().gerbil.block_size, + config.getRawConfig().gerbil.subnet_group + ); + if (!subnet) { + throw new Error("No available subnets remaining in space"); + } + + // replace the last octet with 1 + subnet = + subnet.split(".").slice(0, 3).join(".") + + ".1" + + "/" + + subnet.split("/")[1]; + return { value: subnet, release }; + } catch (e) { + await release(); + throw e; + } } diff --git a/server/private/routers/gerbil/createExitNode.ts b/server/private/routers/gerbil/createExitNode.ts index 818c5f0e1..cfa7c42eb 100644 --- a/server/private/routers/gerbil/createExitNode.ts +++ b/server/private/routers/gerbil/createExitNode.ts @@ -29,37 +29,41 @@ export async function createExitNode( .where(eq(exitNodes.publicKey, publicKey)); let exitNode: ExitNode; if (!exitNodeQuery) { - const address = await getNextAvailableSubnet(); - // 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; - let subEndpoint = ""; - if (config.getRawConfig().gerbil.use_subdomain) { - subEndpoint = await getUniqueExitNodeEndpointName(); + 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; + let subEndpoint = ""; + if (config.getRawConfig().gerbil.use_subdomain) { + subEndpoint = await getUniqueExitNodeEndpointName(); + } + + const exitNodeName = + config.getRawConfig().gerbil.exit_node_name || + `Exit Node ${publicKey.slice(0, 8)}`; + + // create a new exit node + [exitNode] = await db + .insert(exitNodes) + .values({ + publicKey, + endpoint: `${subEndpoint}${subEndpoint != "" ? "." : ""}${config.getRawConfig().gerbil.base_endpoint}`, + address, + listenPort, + online: true, + reachableAt, + name: exitNodeName + }) + .returning() + .execute(); + + logger.info( + `Created new exit node ${exitNode.name} with address ${exitNode.address} and port ${exitNode.listenPort}` + ); + } finally { + await release(); } - - const exitNodeName = - config.getRawConfig().gerbil.exit_node_name || - `Exit Node ${publicKey.slice(0, 8)}`; - - // create a new exit node - [exitNode] = await db - .insert(exitNodes) - .values({ - publicKey, - endpoint: `${subEndpoint}${subEndpoint != "" ? "." : ""}${config.getRawConfig().gerbil.base_endpoint}`, - address, - listenPort, - online: true, - reachableAt, - name: exitNodeName - }) - .returning() - .execute(); - - logger.info( - `Created new exit node ${exitNode.name} with address ${exitNode.address} and port ${exitNode.listenPort}` - ); } else { // update the reachable at [exitNode] = await db diff --git a/server/private/routers/remoteExitNode/createRemoteExitNode.ts b/server/private/routers/remoteExitNode/createRemoteExitNode.ts index bb16f228d..bf86ed107 100644 --- a/server/private/routers/remoteExitNode/createRemoteExitNode.ts +++ b/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,89 +189,106 @@ export async function createRemoteExitNode( ); } - await db.transaction(async (trx) => { - if (!existingExitNode) { - const [res] = await trx - .insert(exitNodes) - .values({ - name: remoteExitNodeId, - address, - endpoint: "", - publicKey: "", - listenPort: 0, - online: false, - type: "remoteExitNode" - }) - .returning(); - existingExitNode = res; - } + // 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) | null = null; + let newExitNodeAddress: string | null = null; + if (!existingExitNode) { + const { value, release } = await getNextAvailableSubnet(); + newExitNodeAddress = value; + releaseSubnetLock = release; + } - if (!existingRemoteExitNode) { - await trx.insert(remoteExitNodes).values({ - remoteExitNodeId: remoteExitNodeId, - secretHash, - dateCreated: moment().toISOString(), - exitNodeId: existingExitNode.exitNodeId - }); - } else { - // update the existing remote exit node - await trx - .update(remoteExitNodes) - .set({ - exitNodeId: existingExitNode.exitNodeId - }) - .where( - eq( - remoteExitNodes.remoteExitNodeId, - existingRemoteExitNode.remoteExitNodeId - ) - ); - } - - if (!existingExitNodeOrg) { - await trx.insert(exitNodeOrgs).values({ - exitNodeId: existingExitNode.exitNodeId, - orgId: orgId - }); - } - - // calculate if the node is in any other of the orgs before we count it as an add to the billing org - if (org.billingOrgId) { - const otherBillingOrgs = await trx - .select() - .from(orgs) - .where( - and( - eq(orgs.billingOrgId, org.billingOrgId), - ne(orgs.orgId, orgId) - ) - ); - - const billingOrgIds = otherBillingOrgs.map((o) => o.orgId); - - const orgsInBillingDomainThatTheNodeIsStillIn = await trx - .select() - .from(exitNodeOrgs) - .where( - and( - eq( - exitNodeOrgs.exitNodeId, - existingExitNode.exitNodeId - ), - inArray(exitNodeOrgs.orgId, billingOrgIds) - ) - ); - - if (orgsInBillingDomainThatTheNodeIsStillIn.length === 0) { - await usageService.add( - orgId, - LimitId.REMOTE_EXIT_NODES, - 1, - trx - ); + try { + await db.transaction(async (trx) => { + if (!existingExitNode) { + const [res] = await trx + .insert(exitNodes) + .values({ + name: remoteExitNodeId, + address: newExitNodeAddress!, + endpoint: "", + publicKey: "", + listenPort: 0, + online: false, + type: "remoteExitNode" + }) + .returning(); + existingExitNode = res; } - } - }); + + if (!existingRemoteExitNode) { + await trx.insert(remoteExitNodes).values({ + remoteExitNodeId: remoteExitNodeId, + secretHash, + dateCreated: moment().toISOString(), + exitNodeId: existingExitNode.exitNodeId + }); + } else { + // update the existing remote exit node + await trx + .update(remoteExitNodes) + .set({ + exitNodeId: existingExitNode.exitNodeId + }) + .where( + eq( + remoteExitNodes.remoteExitNodeId, + existingRemoteExitNode.remoteExitNodeId + ) + ); + } + + if (!existingExitNodeOrg) { + await trx.insert(exitNodeOrgs).values({ + exitNodeId: existingExitNode.exitNodeId, + orgId: orgId + }); + } + + // calculate if the node is in any other of the orgs before we count it as an add to the billing org + if (org.billingOrgId) { + const otherBillingOrgs = await trx + .select() + .from(orgs) + .where( + and( + eq(orgs.billingOrgId, org.billingOrgId), + ne(orgs.orgId, orgId) + ) + ); + + const billingOrgIds = otherBillingOrgs.map((o) => o.orgId); + + const orgsInBillingDomainThatTheNodeIsStillIn = await trx + .select() + .from(exitNodeOrgs) + .where( + and( + eq( + exitNodeOrgs.exitNodeId, + existingExitNode.exitNodeId + ), + inArray(exitNodeOrgs.orgId, billingOrgIds) + ) + ); + + if (orgsInBillingDomainThatTheNodeIsStillIn.length === 0) { + await usageService.add( + orgId, + LimitId.REMOTE_EXIT_NODES, + 1, + trx + ); + } + } + }); + } finally { + await releaseSubnetLock?.(); + } const token = generateSessionToken(); await createRemoteExitNodeSession(token, remoteExitNodeId); diff --git a/server/routers/gerbil/createExitNode.ts b/server/routers/gerbil/createExitNode.ts index bc9650367..9e93cf575 100644 --- a/server/routers/gerbil/createExitNode.ts +++ b/server/routers/gerbil/createExitNode.ts @@ -13,37 +13,41 @@ export async function createExitNode( const [exitNodeQuery] = await db.select().from(exitNodes).limit(1); let exitNode: ExitNode; if (!exitNodeQuery) { - const address = await getNextAvailableSubnet(); - // 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; - let subEndpoint = ""; - if (config.getRawConfig().gerbil.use_subdomain) { - subEndpoint = await getUniqueExitNodeEndpointName(); + 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; + let subEndpoint = ""; + if (config.getRawConfig().gerbil.use_subdomain) { + subEndpoint = await getUniqueExitNodeEndpointName(); + } + + const exitNodeName = + config.getRawConfig().gerbil.exit_node_name || + `Exit Node ${publicKey.slice(0, 8)}`; + + // create a new exit node + [exitNode] = await db + .insert(exitNodes) + .values({ + publicKey, + endpoint: `${subEndpoint}${subEndpoint != "" ? "." : ""}${config.getRawConfig().gerbil.base_endpoint}`, + address, + online: true, + listenPort, + reachableAt, + name: exitNodeName + }) + .returning() + .execute(); + + logger.info( + `Created new exit node ${exitNode.name} with address ${exitNode.address} and port ${exitNode.listenPort}` + ); + } finally { + await release(); } - - const exitNodeName = - config.getRawConfig().gerbil.exit_node_name || - `Exit Node ${publicKey.slice(0, 8)}`; - - // create a new exit node - [exitNode] = await db - .insert(exitNodes) - .values({ - publicKey, - endpoint: `${subEndpoint}${subEndpoint != "" ? "." : ""}${config.getRawConfig().gerbil.base_endpoint}`, - address, - online: true, - listenPort, - reachableAt, - name: exitNodeName - }) - .returning() - .execute(); - - logger.info( - `Created new exit node ${exitNode.name} with address ${exitNode.address} and port ${exitNode.listenPort}` - ); } else { // update the existing exit node [exitNode] = await db From fc54ad49b5fe9af880afec11aebb4a0b49de7aae Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 2 Jul 2026 21:46:44 -0400 Subject: [PATCH 4/8] Fix trial showing --- src/app/[orgId]/settings/(private)/billing/page.tsx | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/app/[orgId]/settings/(private)/billing/page.tsx b/src/app/[orgId]/settings/(private)/billing/page.tsx index aa121838f..0529622e2 100644 --- a/src/app/[orgId]/settings/(private)/billing/page.tsx +++ b/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 From b93d26f09f691827044b90d0a71c5398e73a06ba Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 2 Jul 2026 21:46:53 -0400 Subject: [PATCH 5/8] Fix reading from replicas --- server/private/lib/redis.ts | 59 ++++++++++++++++++++++++++----------- 1 file changed, 41 insertions(+), 18 deletions(-) diff --git a/server/private/lib/redis.ts b/server/private/lib/redis.ts index bda2e6204..7ff24ba93 100644 --- a/server/private/lib/redis.ts +++ b/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.cluster.local -> + // redis-1.redis-headless..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 - // The real replica routing is handled by the StatefulSet headless service - this.readClient = this.writeClient; + const replicaHost = this.getReplicaHost(cfg.host!); + this.readClient = replicaHost + ? 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 { 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); From 1b1fba60f141cc3ef6dcf24efee59dab36291e1b Mon Sep 17 00:00:00 2001 From: Owen Date: Fri, 3 Jul 2026 09:02:08 -0400 Subject: [PATCH 6/8] Make sure to retry rebuilds --- server/lib/dbRetry.ts | 83 +++++++++++++++++++++++++ server/lib/rebuildClientAssociations.ts | 56 ++++++++++++++--- server/lib/rebuildQueue.ts | 35 +++++++++-- server/private/lib/rebuildQueue.ts | 40 ++++++++++-- server/routers/newt/pingAccumulator.ts | 83 +------------------------ 5 files changed, 200 insertions(+), 97 deletions(-) create mode 100644 server/lib/dbRetry.ts diff --git a/server/lib/dbRetry.ts b/server/lib/dbRetry.ts new file mode 100644 index 000000000..7f8d1eea8 --- /dev/null +++ b/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( + operation: () => Promise, + context: string, + maxRetries: number = MAX_RETRIES, + baseDelayMs: number = BASE_DELAY_MS +): Promise { + 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; + } + } +} diff --git a/server/lib/rebuildClientAssociations.ts b/server/lib/rebuildClientAssociations.ts index 150001662..b7647a182 100644 --- a/server/lib/rebuildClientAssociations.ts +++ b/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( - `rebuild-client-associations:site-resource:${siteResource.siteResourceId}`, - () => rebuildClientAssociationsFromSiteResourceImpl(siteResource), - REBUILD_ASSOCIATIONS_LOCK_TTL_MS + // 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 + ), + 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); @@ -1656,10 +1678,17 @@ export async function rebuildClientAssociationsFromClient( await incrementOrgRebuildCount(client.orgId); try { const trx = primaryDb; - return await lockManager.withLock( - `rebuild-client-associations:client:${client.clientId}`, - () => rebuildClientAssociationsFromClientImpl(client, trx), - REBUILD_ASSOCIATIONS_LOCK_TTL_MS + // 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 +1704,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); diff --git a/server/lib/rebuildQueue.ts b/server/lib/rebuildQueue.ts index 07f2cb003..7491f5f52 100644 --- a/server/lib/rebuildQueue.ts +++ b/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,10 +114,29 @@ class InMemoryRebuildQueue implements RebuildQueueManager { `Rebuild queue: completed ${job.type}:${job.id}` ); } catch (err) { - logger.error( - `Rebuild queue: job ${job.type}:${job.id} threw an error:`, - 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 { diff --git a/server/private/lib/rebuildQueue.ts b/server/private/lib/rebuildQueue.ts index b5e112545..87da95f60 100644 --- a/server/private/lib/rebuildQueue.ts +++ b/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,10 +189,33 @@ class RedisRebuildQueue { `Rebuild queue: completed ${job.type}:${job.id}` ); } catch (err) { - logger.error( - `Rebuild queue: job ${job.type}:${job.id} threw an error:`, - 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 + ); + } } } }, diff --git a/server/routers/newt/pingAccumulator.ts b/server/routers/newt/pingAccumulator.ts index be9cd4972..f82190790 100644 --- a/server/routers/newt/pingAccumulator.ts +++ b/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 { } // ── Retry / Error 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( - operation: () => Promise, - context: string -): Promise { - 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; -} +// See @server/lib/dbRetry for the shared withRetry/isTransientError helpers. // ── Lifecycle ────────────────────────────────────────────────────────── From b399d2a291423f535a3138124575e73eeda535a1 Mon Sep 17 00:00:00 2001 From: Owen Date: Fri, 3 Jul 2026 10:23:32 -0400 Subject: [PATCH 7/8] Add some retry and database confict mitigation --- server/lib/rebuildClientAssociations.ts | 257 ++++++++++++---------- server/routers/newt/buildConfiguration.ts | 4 +- server/routers/olm/buildConfiguration.ts | 2 +- 3 files changed, 145 insertions(+), 118 deletions(-) diff --git a/server/lib/rebuildClientAssociations.ts b/server/lib/rebuildClientAssociations.ts index b7647a182..abad6b7aa 100644 --- a/server/lib/rebuildClientAssociations.ts +++ b/server/lib/rebuildClientAssociations.ts @@ -485,6 +485,7 @@ async function rebuildClientAssociationsFromSiteResourceImpl( await trx .insert(clientSiteResourcesAssociationsCache) .values(clientSiteResourcesToInsert) + .onConflictDoNothing() .returning(); logger.debug( `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteResourceId=${siteResource.siteResourceId} inserted clientSiteResource associations` @@ -532,121 +533,141 @@ async function rebuildClientAssociationsFromSiteResourceImpl( for (const site of sitesToProcess) { const siteId = site.siteId; - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] processing siteId=${siteId} for siteResourceId=${siteResource.siteResourceId}` - ); + try { + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] processing siteId=${siteId} for siteResourceId=${siteResource.siteResourceId}` + ); - const existingClientSites = await trx - .select({ - clientId: clientSitesAssociationsCache.clientId - }) - .from(clientSitesAssociationsCache) - .where(eq(clientSitesAssociationsCache.siteId, siteId)); + const existingClientSites = await trx + .select({ + clientId: clientSitesAssociationsCache.clientId + }) + .from(clientSitesAssociationsCache) + .where(eq(clientSitesAssociationsCache.siteId, siteId)); - const existingClientSiteIds = existingClientSites.map( - (row) => row.clientId - ); + const existingClientSiteIds = existingClientSites.map( + (row) => row.clientId + ); - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} existingClientSiteIds=[${existingClientSiteIds.join(", ")}]` - ); + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} existingClientSiteIds=[${existingClientSiteIds.join(", ")}]` + ); - // Get full client details for existing clients (needed for sending delete messages) - const existingClients = - existingClientSiteIds.length > 0 - ? await trx - .select({ - clientId: clients.clientId, - pubKey: clients.pubKey, - subnet: clients.subnet - }) - .from(clients) - .where(inArray(clients.clientId, existingClientSiteIds)) + // Get full client details for existing clients (needed for sending delete messages) + const existingClients = + existingClientSiteIds.length > 0 + ? await trx + .select({ + clientId: clients.clientId, + pubKey: clients.pubKey, + subnet: clients.subnet + }) + .from(clients) + .where( + inArray(clients.clientId, existingClientSiteIds) + ) + : []; + + const otherResourceClientIds = + clientsFromOtherResourcesBySite.get(siteId) ?? + new Set(); + + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} otherResourceClientIds=[${[...otherResourceClientIds].join(", ")}] mergedAllClientIds=[${mergedAllClientIds.join(", ")}]` + ); + + // Expected clients from this resource are site-scoped: if this site is + // no longer attached to the resource, the expected set is empty. + const expectedClientIdsForSite = currentSiteIdSet.has(siteId) + ? mergedAllClientIds : []; - const otherResourceClientIds = - clientsFromOtherResourcesBySite.get(siteId) ?? new Set(); - - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} otherResourceClientIds=[${[...otherResourceClientIds].join(", ")}] mergedAllClientIds=[${mergedAllClientIds.join(", ")}]` - ); - - // Expected clients from this resource are site-scoped: if this site is - // no longer attached to the resource, the expected set is empty. - const expectedClientIdsForSite = currentSiteIdSet.has(siteId) - ? mergedAllClientIds - : []; - - const clientSitesToAdd = expectedClientIdsForSite.filter( - (clientId) => - !existingClientSiteIds.includes(clientId) && - !otherResourceClientIds.has(clientId) // dont add if already connected via another site resource - ); - - const clientSitesToInsert = clientSitesToAdd.map((clientId) => ({ - clientId, - siteId - })); - - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toAdd=[${clientSitesToAdd.join(", ")}]` - ); - - if (clientSitesToInsert.length > 0) { - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserting ${clientSitesToInsert.length} clientSite association(s)` + const clientSitesToAdd = expectedClientIdsForSite.filter( + (clientId) => + !existingClientSiteIds.includes(clientId) && + !otherResourceClientIds.has(clientId) // dont add if already connected via another site resource ); - await trx - .insert(clientSitesAssociationsCache) - .values(clientSitesToInsert) - .returning(); - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserted clientSite associations` - ); - } else { - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} no clientSite associations to insert` - ); - } - // Now remove any client-site associations that should no longer exist - const clientSitesToRemove = existingClientSiteIds.filter( - (clientId) => - !expectedClientIdsForSite.includes(clientId) && - !otherResourceClientIds.has(clientId) // dont remove if there is still another connection for another site resource - ); + const clientSitesToInsert = clientSitesToAdd.map((clientId) => ({ + clientId, + siteId + })); - logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toRemove=[${clientSitesToRemove.join(", ")}]` - ); - - if (clientSitesToRemove.length > 0) { logger.debug( - `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} deleting ${clientSitesToRemove.length} clientSite association(s)` + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toAdd=[${clientSitesToAdd.join(", ")}]` ); - await trx - .delete(clientSitesAssociationsCache) - .where( - and( - eq(clientSitesAssociationsCache.siteId, siteId), - inArray( - clientSitesAssociationsCache.clientId, - clientSitesToRemove - ) - ) + + if (clientSitesToInsert.length > 0) { + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserting ${clientSitesToInsert.length} clientSite association(s)` ); - } + await trx + .insert(clientSitesAssociationsCache) + .values(clientSitesToInsert) + .onConflictDoNothing() + .returning(); + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} inserted clientSite associations` + ); + } else { + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} no clientSite associations to insert` + ); + } - // Now handle the messages to add/remove peers on both the newt and olm sides - await handleMessagesForSiteClients( - site, - siteId, - mergedAllClients, - existingClients, - clientSitesToAdd, - clientSitesToRemove, - trx - ); + // Now remove any client-site associations that should no longer exist + const clientSitesToRemove = existingClientSiteIds.filter( + (clientId) => + !expectedClientIdsForSite.includes(clientId) && + !otherResourceClientIds.has(clientId) // dont remove if there is still another connection for another site resource + ); + + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} clientSites toRemove=[${clientSitesToRemove.join(", ")}]` + ); + + if (clientSitesToRemove.length > 0) { + logger.debug( + `rebuildClientAssociations: [rebuildClientAssociationsFromSiteResource] siteId=${siteId} deleting ${clientSitesToRemove.length} clientSite association(s)` + ); + await trx + .delete(clientSitesAssociationsCache) + .where( + and( + eq(clientSitesAssociationsCache.siteId, siteId), + inArray( + clientSitesAssociationsCache.clientId, + clientSitesToRemove + ) + ) + ); + } + + // Now handle the messages to add/remove peers on both the newt and olm sides + await handleMessagesForSiteClients( + site, + siteId, + mergedAllClients, + existingClients, + clientSitesToAdd, + 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 @@ -939,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; } @@ -1866,12 +1887,15 @@ async function rebuildClientAssociationsFromClientImpl( // Insert new associations if (resourcesToAdd.length > 0) { - await trx.insert(clientSiteResourcesAssociationsCache).values( - resourcesToAdd.map((siteResourceId) => ({ - clientId: client.clientId, - siteResourceId - })) - ); + await trx + .insert(clientSiteResourcesAssociationsCache) + .values( + resourcesToAdd.map((siteResourceId) => ({ + clientId: client.clientId, + siteResourceId + })) + ) + .onConflictDoNothing(); } // Remove old associations @@ -1909,12 +1933,15 @@ async function rebuildClientAssociationsFromClientImpl( // Insert new site associations if (sitesToAdd.length > 0) { - await trx.insert(clientSitesAssociationsCache).values( - sitesToAdd.map((siteId) => ({ - clientId: client.clientId, - siteId - })) - ); + await trx + .insert(clientSitesAssociationsCache) + .values( + sitesToAdd.map((siteId) => ({ + clientId: client.clientId, + siteId + })) + ) + .onConflictDoNothing(); } // Remove old site associations diff --git a/server/routers/newt/buildConfiguration.ts b/server/routers/newt/buildConfiguration.ts index 5083a6c56..5b7f211ae 100644 --- a/server/routers/newt/buildConfiguration.ts +++ b/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; diff --git a/server/routers/olm/buildConfiguration.ts b/server/routers/olm/buildConfiguration.ts index 41bb6d60d..f72731c92 100644 --- a/server/routers/olm/buildConfiguration.ts +++ b/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; } From 440ebfe08e1ecf2bc0f31f2e48a9bb12ade38e08 Mon Sep 17 00:00:00 2001 From: Owen Date: Fri, 3 Jul 2026 10:26:13 -0400 Subject: [PATCH 8/8] Clarify error messages --- server/routers/newt/handleNewtDisconnectingMessage.ts | 2 +- server/routers/olm/handleOlmDisconnectingMessage.ts | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/server/routers/newt/handleNewtDisconnectingMessage.ts b/server/routers/newt/handleNewtDisconnectingMessage.ts index a2b963fc9..afc208ac8 100644 --- a/server/routers/newt/handleNewtDisconnectingMessage.ts +++ b/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); } }; diff --git a/server/routers/olm/handleOlmDisconnectingMessage.ts b/server/routers/olm/handleOlmDisconnectingMessage.ts index ecd101724..241183497 100644 --- a/server/routers/olm/handleOlmDisconnectingMessage.ts +++ b/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); } };