Merge branch 'main' into dev

server/lib/deleteOrg.childTest.ts

@@ -0,0 +1,139 @@
+// Test child for deleteOrg.test.ts. Boots the real sqlite driver against a
+// fresh database in the temp workdir (cwd), seeds an org with two wireguard
+// sites and two newt sites, runs deleteOrgById against a local HTTP stand-in
+// for the exit node, and prints one JSON line of observations.
+// The app boots with `import "./extendZod"` before anything else; zod
+// schemas in the import chain use .openapi(), so do the same here.
+import "@server/extendZod";
+import http from "http";
+import { deleteOrgById } from "@server/lib/deleteOrg";
+import {
+    db,
+    exitNodes,
+    orgs,
+    sites,
+    newts,
+    newtSessions,
+    clientSitesAssociationsCache
+} from "@server/db";
+import { count, eq } from "drizzle-orm";
+
+const peerRemovals: string[] = [];
+const server = http.createServer((req, res) => {
+    let body = "";
+    req.on("data", (c) => (body += c));
+    req.on("end", () => {
+        if (req.method === "DELETE") {
+            if (!req.url) {
+                res.writeHead(200, { "Content-Type": "application/json" });
+                res.end(JSON.stringify({ status: "ok" }));
+                return;
+            }
+            const url = new URL(req.url, "http://localhost");
+            const pubKey = url.searchParams.get("public_key");
+            if (pubKey !== null) {
+                peerRemovals.push(pubKey);
+            }
+        }
+        res.writeHead(200, { "Content-Type": "application/json" });
+        res.end(JSON.stringify({ status: "ok" }));
+    });
+});
+
+server.listen(0, "127.0.0.1", async () => {
+    const addr = server.address();
+    if (addr === null || typeof addr === "string") {
+        throw new Error("failed to bind the stub exit node");
+    }
+    const port = addr.port;
+    try {
+        await db.insert(exitNodes).values({
+            exitNodeId: 1,
+            name: "en",
+            address: "10.0.0.1",
+            endpoint: "10.0.0.1:51820",
+            publicKey: "enKey",
+            listenPort: 51820,
+            reachableAt: "http://127.0.0.1:" + port,
+            type: "wireguard"
+        });
+        await db.insert(orgs).values({
+            orgId: "org1",
+            name: "Org",
+            settingsJitModeLimit: 1
+        });
+        for (let i = 1; i <= 2; i++) {
+            await db.insert(sites).values({
+                siteId: i,
+                orgId: "org1",
+                niceId: String(i),
+                name: "wg" + i,
+                pubKey: "wgKey" + i,
+                exitNodeId: 1,
+                type: "wireguard"
+            });
+        }
+        for (let i = 3; i <= 4; i++) {
+            await db.insert(sites).values({
+                siteId: i,
+                orgId: "org1",
+                niceId: String(i),
+                name: "newt" + i,
+                pubKey: "newtPub" + i,
+                publicKey: "newtSession" + i,
+                type: "newt"
+            });
+            await db.insert(newts).values({
+                newtId: "newtId" + i,
+                secretHash: "h" + i,
+                dateCreated: "2026-01-01",
+                siteId: i
+            });
+            await db.insert(newtSessions).values({
+                sessionId: "sess" + i,
+                newtId: "newtId" + i,
+                expiresAt: 4102444800
+            });
+        }
+        // A stale association-cache row pinned to one of the newt sites, the
+        // kind cleanupSiteAssociations exists to clear.
+        await db.insert(clientSitesAssociationsCache).values({
+            clientId: 99,
+            siteId: 3,
+            isRelayed: false
+        });
+
+        const result = await deleteOrgById("org1");
+
+        // Read the leftovers through the query builder, not db.get(): the
+        // pg build type-checks this file against the pg driver too, and
+        // .get() exists only on the sqlite driver.
+        const [sitesRow] = await db.select({ count: count() }).from(sites);
+        const [newtsRow] = await db.select({ count: count() }).from(newts);
+        const [sessionsRow] = await db
+            .select({ count: count() })
+            .from(newtSessions);
+        const [assocRow] = await db
+            .select({ count: count() })
+            .from(clientSitesAssociationsCache)
+            .where(eq(clientSitesAssociationsCache.siteId, 3));
+
+        console.log(
+            "OBSERVATIONS " +
+                JSON.stringify({
+                    peerRemovals: peerRemovals.slice().sort(),
+                    terminatedNewtIds: result.deletedNewtIds.slice().sort(),
+                    sitesLeft: sitesRow.count,
+                    newtsLeft: newtsRow.count,
+                    sessionsLeft: sessionsRow.count,
+                    associationRowsLeft: assocRow.count
+                })
+        );
+        server.close();
+        process.exit(0);
+    } catch (e: any) {
+        console.error("CHILD_FAIL", e && e.message ? e.message : String(e));
+        server.close();
+        process.exit(1);
+    }
+});

server/lib/deleteOrg.test.ts

@@ -0,0 +1,185 @@
+import { execFileSync } from "child_process";
+import fs from "fs";
+import os from "os";
+import path from "path";
+import { fileURLToPath } from "url";
+import { assertEquals } from "@test/assert";
+
+const here = path.dirname(fileURLToPath(import.meta.url));
+const repoRoot = path.resolve(here, "..", "..");
+
+const CHILD_SCRIPT_PATH = path.join(
+    repoRoot,
+    "server",
+    "lib",
+    "deleteOrg.childTest.ts"
+);
+
+function runChild(workdir: string): {
+    exitCode: number;
+    observations: any;
+    output: string;
+} {
+    const tsconfig = ["tsconfig.json", "tsconfig.oss.json"]
+        .map((file) => path.join(repoRoot, file))
+        .find((file) => fs.existsSync(file));
+    if (!tsconfig) {
+        throw new Error("No tsconfig found for @server path aliases");
+    }
+    const tsxCli = path.join(
+        repoRoot,
+        "node_modules",
+        "tsx",
+        "dist",
+        "cli.mjs"
+    );
+    if (!fs.existsSync(tsxCli)) {
+        throw new Error("tsx is not installed; run npm ci first");
+    }
+    try {
+        const output = execFileSync(
+            process.execPath,
+            [tsxCli, "--tsconfig", tsconfig, CHILD_SCRIPT_PATH],
+            {
+                cwd: workdir,
+                timeout: 120000,
+                encoding: "utf8",
+                env: { ...process.env, NODE_ENV: "test" }
+            }
+        );
+        const m = output.match(/OBSERVATIONS (.*)/);
+        if (!m) {
+            throw new Error("child did not print observations: " + output);
+        }
+        return { exitCode: 0, observations: JSON.parse(m[1]), output };
+    } catch (error: any) {
+        const output = String(error?.stdout ?? "");
+        return { exitCode: 1, observations: null, output };
+    }
+}
+
+function createTestEnvironment(): string {
+    for (const generated of ["server/build.ts", "server/db/index.ts"]) {
+        if (!fs.existsSync(path.join(repoRoot, generated))) {
+            throw new Error(
+                `Missing ${generated}; run npm run set:oss && npm run set:sqlite first`
+            );
+        }
+    }
+    const workdir = fs.mkdtempSync(
+        path.join(os.tmpdir(), "pangolin-org-cleanup-")
+    );
+    fs.mkdirSync(path.join(workdir, "config", "db"), { recursive: true });
+    fs.copyFileSync(
+        path.join(repoRoot, "config", "config.example.yml"),
+        path.join(workdir, "config", "config.yml")
+    );
+    fs.symlinkSync(
+        path.join(repoRoot, "server"),
+        path.join(workdir, "server"),
+        process.platform === "win32" ? "junction" : "dir"
+    );
+    fs.symlinkSync(
+        path.join(repoRoot, "test"),
+        path.join(workdir, "test"),
+        process.platform === "win32" ? "junction" : "dir"
+    );
+    fs.symlinkSync(
+        path.join(repoRoot, "node_modules"),
+        path.join(workdir, "node_modules"),
+        process.platform === "win32" ? "junction" : "dir"
+    );
+    return workdir;
+}
+
+function applyMigrations(workdir: string) {
+    const tsconfig = ["tsconfig.json", "tsconfig.oss.json"]
+        .map((file) => path.join(repoRoot, file))
+        .find((file) => fs.existsSync(file));
+    if (!tsconfig) {
+        throw new Error("No tsconfig found for @server path aliases");
+    }
+    const tsxCli = path.join(
+        repoRoot,
+        "node_modules",
+        "tsx",
+        "dist",
+        "cli.mjs"
+    );
+    // CI applies migrations with `npm run db:push` before booting the app;
+    // server/db/migrate.ts is the drizzle migrator that script invokes.
+    const migrationsScript = path.join(repoRoot, "server", "db", "migrate.ts");
+    execFileSync(
+        process.execPath,
+        [tsxCli, "--tsconfig", tsconfig, migrationsScript],
+        {
+            cwd: workdir,
+            timeout: 120000,
+            encoding: "utf8",
+            env: { ...process.env, NODE_ENV: "test" }
+        }
+    );
+}
+
+function testOrgCleanupCoversEverySite() {
+    console.log("Running deleteOrgById per-site cleanup test...");
+    const workdir = createTestEnvironment();
+    applyMigrations(workdir);
+
+    const { exitCode, observations, output } = runChild(workdir);
+    if (exitCode !== 0) {
+        throw new Error("deleteOrgById child run failed: " + output);
+    }
+
+    // The exact bug from the issue: with a single-row sites fetch, only one
+    // of the four sites gets its side effects. All wireguard peers must be
+    // removed from the exit node...
+    assertEquals(
+        JSON.stringify(observations.peerRemovals),
+        JSON.stringify(["wgKey1", "wgKey2"]),
+        "deleteOrgById must remove every wireguard site's peer from its exit node"
+    );
+    // ...and every newt connector must be collected for the terminate message.
+    assertEquals(
+        JSON.stringify(observations.terminatedNewtIds),
+        JSON.stringify(["newtId3", "newtId4"]),
+        "deleteOrgById must collect every newt id so all connectors get the terminate message"
+    );
+    // DB rows must still cascade to nothing left.
+    assertEquals(observations.sitesLeft, 0, "all site rows must be gone");
+    assertEquals(observations.newtsLeft, 0, "all newt rows must be gone");
+    assertEquals(
+        observations.sessionsLeft,
+        0,
+        "all newt session rows must be gone"
+    );
+
+    // The gap the issue calls out separately: cleanupSiteAssociations must
+    // run for newt sites (deleteSite.ts already does this), so the stale
+    // association cache rows for these sites must be cleared.
+    assertEquals(
+        observations.associationRowsLeft,
+        0,
+        "newt site cleanup must clear clientSitesAssociationsCache rows"
+    );
+
+    console.log("deleteOrgById per-site cleanup test passed");
+}
+
+function main() {
+    let failed = 0;
+    for (const test of [testOrgCleanupCoversEverySite]) {
+        try {
+            test();
+        } catch (error) {
+            failed++;
+            console.error(`FAIL: ${(error as Error).message}`);
+        }
+    }
+    if (failed > 0) {
+        process.exit(1);
+    }
+    console.log("deleteOrg tests complete");
+}
+
+main();

server/lib/deleteOrg.ts

@@ -23,6 +23,7 @@ import { sendToClient } from "#dynamic/routers/ws";
 import { deletePeer } from "@server/routers/gerbil/peers";
 import { OlmErrorCodes } from "@server/routers/olm/error";
 import { sendTerminateClient } from "@server/routers/client/terminate";
+import { cleanupSiteAssociations } from "@server/lib/rebuildClientAssociations";
 import { usageService } from "./billing/usageService";
 import { LimitId } from "./billing";
 
@@ -55,8 +56,7 @@ export async function deleteOrgById(
     const orgSites = await db
         .select()
         .from(sites)
-        .where(eq(sites.orgId, orgId))
-        .limit(1);
+        .where(eq(sites.orgId, orgId));
 
     const orgClients = await db
         .select()
@@ -70,23 +70,31 @@ export async function deleteOrgById(
     let siteCount: number | null = null;
     let userCount: number | null = null;
     let remoteExitNodeCount: number | null = null;
+    // billingOrgId has no FK, so the billing org may already be gone (e.g.
+    // a partially failed account deletion). Writing usage for it would
+    // violate the usage FK and abort the transaction.
+    let billingOrgExists = false;
 
     await db.transaction(async (trx) => {
         for (const site of orgSites) {
-            if (site.pubKey) {
-                if (site.type == "wireguard") {
+            if (site.type == "wireguard") {
+                if (site.pubKey) {
                     await deletePeer(site.exitNodeId!, site.pubKey);
-                } else if (site.type == "newt") {
-                    const [deletedNewt] = await trx
-                        .delete(newts)
-                        .where(eq(newts.siteId, site.siteId))
-                        .returning();
-                    if (deletedNewt) {
-                        deletedNewtIds.push(deletedNewt.newtId);
-                        await trx
-                            .delete(newtSessions)
-                            .where(eq(newtSessions.newtId, deletedNewt.newtId));
-                    }
+                }
+            } else if (site.type == "newt") {
+                // Same per-site cleanup deleteSite.ts performs, so that
+                // removing an org does not leave other orgs' clients
+                // holding stale associations for these sites.
+                await cleanupSiteAssociations(site, trx);
+                const [deletedNewt] = await trx
+                    .delete(newts)
+                    .where(eq(newts.siteId, site.siteId))
+                    .returning();
+                if (deletedNewt) {
+                    deletedNewtIds.push(deletedNewt.newtId);
+                    await trx
+                        .delete(newtSessions)
+                        .where(eq(newtSessions.newtId, deletedNewt.newtId));
                 }
             }
             logger.info(`Deleting site ${site.siteId}`);
@@ -157,7 +165,17 @@ export async function deleteOrgById(
                 .where(inArray(domains.domainId, domainIdsToDelete));
         }
 
-        await usageService.add(orgId, LimitId.ORGANIZATIONS, -1, trx); // here we are decreasing the org count BEFORE deleting the org because we need to still be able to get the org to get the billing org inside of here
+        const [billingOrg] = org.billingOrgId
+            ? await trx
+                  .select({ orgId: orgs.orgId })
+                  .from(orgs)
+                  .where(eq(orgs.orgId, org.billingOrgId))
+                  .limit(1)
+            : [org];
+        billingOrgExists = !!billingOrg;
+        if (billingOrgExists) {
+            await usageService.add(orgId, LimitId.ORGANIZATIONS, -1, trx); // here we are decreasing the org count BEFORE deleting the org because we need to still be able to get the org to get the billing org inside of here
+        }
 
         await trx.delete(orgs).where(eq(orgs.orgId, orgId));
 
@@ -201,7 +219,11 @@ export async function deleteOrgById(
         }
     });
 
-    if (org.billingOrgId) {
+    if (
+        org.billingOrgId &&
+        org.billingOrgId !== orgId &&
+        billingOrgExists
+    ) {
         usageService.updateCount(
             org.billingOrgId,
             LimitId.DOMAINS,

server/lib/exitNodes/getUniqueSubnetForExitNode.ts

@@ -1,41 +1,62 @@
-import { db, ExitNode, Transaction, sites, clients } from "@server/db";
+import { ExitNode, primaryDb, sites, clients } from "@server/db";
 import { eq } from "drizzle-orm";
 import config from "@server/lib/config";
 import { findNextAvailableCidr } from "@server/lib/ip";
 import { lockManager } from "#dynamic/lib/lock";
 
+/**
+ * Picks the next free site/client subnet on the given exit node. This acquires
+ * a per-exit-node lock that the caller MUST release (via the returned
+ * `release`) only after the chosen subnet has been persisted to the database
+ * (or the operation has failed) — otherwise a concurrent caller on another
+ * node could pick the same subnet. If no subnet is available the lock is
+ * released before returning and `value` is null.
+ */
 export async function getUniqueSubnetForExitNode(
-    exitNode: ExitNode,
-    trx: Transaction | typeof db = db
-): Promise<string | null> {
+    exitNode: ExitNode
+): Promise<{ value: string | null; release: () => Promise<void> }> {
     const lockKey = `subnet-allocation:${exitNode.exitNodeId}`;
+    const acquired = await lockManager.acquireLockWithRetry(lockKey, 6000);
+    if (!acquired) {
+        throw new Error(`Failed to acquire lock: ${lockKey}`);
+    }
+    const release = () => lockManager.releaseLock(lockKey, acquired);
 
-    return await lockManager.withLock(
-        lockKey,
-        async () => {
-            const [sitesQuery, clientsQuery] = await Promise.all([
-                trx
-                    .select({ subnet: sites.exitNodeSubnet })
-                    .from(sites)
-                    .where(eq(sites.exitNodeId, exitNode.exitNodeId)),
-                trx
-                    .select({ subnet: clients.exitNodeSubnet })
-                    .from(clients)
-                    .where(eq(clients.exitNodeId, exitNode.exitNodeId))
-            ]);
+    try {
+        // read from the primary so we see subnets committed by the previous lock holder
+        const [sitesQuery, clientsQuery] = await Promise.all([
+            primaryDb
+                .select({ subnet: sites.exitNodeSubnet })
+                .from(sites)
+                .where(eq(sites.exitNodeId, exitNode.exitNodeId)),
+            primaryDb
+                .select({ subnet: clients.exitNodeSubnet })
+                .from(clients)
+                .where(eq(clients.exitNodeId, exitNode.exitNodeId))
+        ]);
 
-            const blockSize = config.getRawConfig().gerbil.site_block_size;
-            const subnets = [...sitesQuery, ...clientsQuery]
-                .map((row) => row.subnet)
-                .filter(
-                    (subnet): subnet is string =>
-                        !!subnet &&
-                        /^(\d{1,3}\.){3}\d{1,3}\/\d{1,2}$/.test(subnet)
-                );
-            subnets.push(exitNode.address.replace(/\/\d+$/, `/${blockSize}`));
+        const blockSize = config.getRawConfig().gerbil.site_block_size;
+        const subnets = [...sitesQuery, ...clientsQuery]
+            .map((row) => row.subnet)
+            .filter(
+                (subnet): subnet is string =>
+                    !!subnet && /^(\d{1,3}\.){3}\d{1,3}\/\d{1,2}$/.test(subnet)
+            );
+        subnets.push(exitNode.address.replace(/\/\d+$/, `/${blockSize}`));
 
-            return findNextAvailableCidr(subnets, blockSize, exitNode.address);
-        },
-        5000 // 5 second lock TTL - subnet allocation should be quick
-    );
+        const subnet = findNextAvailableCidr(
+            subnets,
+            blockSize,
+            exitNode.address
+        );
+        if (!subnet) {
+            await release();
+            return { value: null, release: async () => {} };
+        }
+
+        return { value: subnet, release };
+    } catch (e) {
+        await release();
+        throw e;
+    }
 }

server/private/lib/certificates/index.ts

@@ -15,6 +15,23 @@ import logger from "@server/logger";
 import { privateConfig } from "#private/lib/config";
 import { acmeClientManager } from "./acme-client";
 import { jobScheduler } from "./scheduler";
+import { LeaderElection } from "#private/lib/leaderElection";
+
+// Only one instance sharing this Redis talks to the ACME server at a time.
+// Every instance runs the election; whichever holds the lease runs issuance
+// and renewal, and another takes over if it goes away.
+const acmeLeaderElection = new LeaderElection("acme-client", {
+    onElected: async () => {
+        logger.info("Starting certificate management server...");
+
+        // Initialize ACME client
+        await acmeClientManager.initialize();
+
+        // Start certificate issuance/renewal jobs
+        await jobScheduler.start();
+    },
+    onDemoted: () => jobScheduler.stopCertJobs()
+});
 
 export async function startCertificateManager() {
     const acmeConfig = privateConfig.getRawPrivateConfig().acme;
@@ -23,13 +40,7 @@ export async function startCertificateManager() {
         acmeConfig.cert_mode === "pangolin" &&
         acmeConfig.enable_acme_client
     ) {
-        logger.info("Starting certificate management server...");
-
-        // Initialize ACME client
-        await acmeClientManager.initialize();
-
-        // Start certificate issuance/renewal jobs
-        await jobScheduler.start();
+        acmeLeaderElection.start();
     }
 
     if (privateConfig.getRawPrivateConfig().flags.use_pangolin_dns) {
@@ -41,5 +52,6 @@ export async function startCertificateManager() {
 }
 
 export async function stopCertificateManager() {
+    await acmeLeaderElection.stop();
     await jobScheduler.stop();
 }

server/private/lib/certificates/scheduler.ts

@@ -173,13 +173,21 @@ export class JobScheduler {
         logger.info("DNS validation job scheduler started successfully");
     }
 
-    async stop(): Promise<void> {
+    // Stops only issuance/renewal (e.g. when this instance stops being the
+    // ACME leader) and leaves the DNS jobs running. A batch already in
+    // flight finishes on its own; the per-domain cert locks keep it from
+    // overlapping with the new leader's work on the same certificate.
+    async stopCertJobs(): Promise<void> {
         if (this.certRunning) {
             logger.info("Stopping certificate job scheduler");
             this.certRunning = false;
             this.certIntervals.forEach((interval) => clearInterval(interval));
             this.certIntervals = [];
         }
+    }
+
+    async stop(): Promise<void> {
+        await this.stopCertJobs();
 
         if (this.dnsRunning) {
             logger.info("Stopping DNS validation job scheduler");

server/private/lib/leaderElection.ts

@@ -0,0 +1,259 @@
+/*
+ * This file is part of a proprietary work.
+ *
+ * Copyright (c) 2025-2026 Fossorial, Inc.
+ * All rights reserved.
+ *
+ * This file is licensed under the Fossorial Commercial License.
+ * You may not use this file except in compliance with the License.
+ * Unauthorized use, copying, modification, or distribution is strictly prohibited.
+ *
+ * This file is not licensed under the AGPLv3.
+ */
+
+import { v4 as uuidv4 } from "uuid";
+import redisManager from "#private/lib/redis";
+import { withTimeout } from "@server/lib/retry";
+import logger from "@server/logger";
+
+// How long a lease lasts without being renewed. If the leader dies, the
+// role sits empty for at most this long before a follower can take it.
+const LEASE_TTL_MS = 30_000;
+
+// How often the leader renews its lease and a follower tries to take it.
+const TICK_INTERVAL_MS = 10_000;
+
+// A leader that can't confirm its lease within this time steps down. Even
+// when the last successful renewal was just before a tick, it stops within
+// TICK_INTERVAL_MS + REDIS_TIMEOUT_MS (15s), well before the lease can run
+// out (LEASE_TTL_MS) and another instance can take it.
+const REDIS_TIMEOUT_MS = 5_000;
+
+const RENEW_SCRIPT = `
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('PEXPIRE', KEYS[1], ARGV[2])
+else
+    return 0
+end
+`;
+
+const RELEASE_SCRIPT = `
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('DEL', KEYS[1])
+else
+    return 0
+end
+`;
+
+type LeaderCallbacks = {
+    // Runs once each time this instance becomes leader. If it throws, the
+    // lease is given up and the next tick tries again (here or elsewhere).
+    onElected: () => Promise<void>;
+    // Runs when leadership is lost or given up. Must be safe to call even if
+    // onElected only partly finished.
+    onDemoted: () => Promise<void>;
+};
+
+/**
+ * Makes sure only one Pangolin instance sharing a Redis is running a given
+ * role at a time.
+ *
+ * Leadership is a Redis key, set with NX and a TTL, whose value is this
+ * instance's token. The leader renews it every tick and steps down as soon
+ * as a renewal fails or times out, so it has stopped before the lease
+ * can run out and another instance can take over.
+ *
+ * Talks to the Redis client directly rather than through lockManager:
+ * lockManager falls back to in-process locks while Redis is unreachable,
+ * which would let every instance become leader during an outage. Here an
+ * unreachable Redis means no instance is leader until it comes back.
+ *
+ * With Redis disabled, the deployment is assumed to be a single instance
+ * (Redis is required to run more than one), so this instance is always the
+ * leader.
+ */
+export class LeaderElection {
+    private readonly key: string;
+    private readonly callbacks: LeaderCallbacks;
+    private token: string | null = null;
+    private timer: NodeJS.Timeout | null = null;
+    private leader = false;
+    private running = false;
+
+    constructor(name: string, callbacks: LeaderCallbacks) {
+        this.key = `leader:${name}`;
+        this.callbacks = callbacks;
+    }
+
+    start(): void {
+        if (this.running) {
+            return;
+        }
+        this.running = true;
+
+        if (!redisManager.getWriteClient()) {
+            logger.info(
+                `Redis is disabled - assuming a single instance and taking ${this.key}`
+            );
+            void this.becomeLeader();
+            return;
+        }
+
+        logger.info(`Leader election started for ${this.key}`);
+        void this.tick();
+    }
+
+    async stop(): Promise<void> {
+        if (!this.running) {
+            return;
+        }
+        this.running = false;
+
+        if (this.timer) {
+            clearTimeout(this.timer);
+            this.timer = null;
+        }
+
+        await this.stepDown("shutting down");
+    }
+
+    isLeader(): boolean {
+        return this.leader;
+    }
+
+    // Ticks are chained with setTimeout rather than setInterval so a slow
+    // acquire/renew can never overlap with the next one.
+    private async tick(): Promise<void> {
+        if (!this.running) {
+            return;
+        }
+
+        try {
+            if (this.leader) {
+                await this.renew();
+            } else {
+                await this.tryAcquire();
+            }
+        } catch (error) {
+            logger.error(`Leader election error for ${this.key}:`, error);
+            await this.stepDown("leader election error");
+        }
+
+        if (this.running) {
+            this.timer = setTimeout(() => void this.tick(), TICK_INTERVAL_MS);
+        }
+    }
+
+    private async tryAcquire(): Promise<void> {
+        const client = redisManager.getWriteClient()!;
+        const token = uuidv4();
+
+        const result = await withTimeout(
+            client.set(this.key, token, "PX", LEASE_TTL_MS, "NX"),
+            REDIS_TIMEOUT_MS,
+            `acquire ${this.key}`
+        );
+
+        if (result !== "OK") {
+            logger.debug(
+                `Not leader for ${this.key} - another instance holds the lease`
+            );
+            return;
+        }
+
+        this.token = token;
+
+        // stop() may have run while the SET was in flight.
+        if (!this.running) {
+            await this.release();
+            return;
+        }
+
+        await this.becomeLeader();
+    }
+
+    private async renew(): Promise<void> {
+        if (!this.token) {
+            throw new Error("leader has no lease token");
+        }
+
+        const client = redisManager.getWriteClient()!;
+        const renewed = (await withTimeout(
+            client.eval(
+                RENEW_SCRIPT,
+                1,
+                this.key,
+                this.token,
+                LEASE_TTL_MS.toString()
+            ),
+            REDIS_TIMEOUT_MS,
+            `renew ${this.key}`
+        )) as number;
+
+        if (renewed !== 1) {
+            throw new Error("lease is no longer held by this instance");
+        }
+    }
+
+    private async becomeLeader(): Promise<void> {
+        this.leader = true;
+        logger.info(`Elected leader for ${this.key}`);
+
+        try {
+            await this.callbacks.onElected();
+        } catch (error) {
+            logger.error(
+                `Failed to start ${this.key} after being elected leader:`,
+                error
+            );
+            // Without Redis there's nobody to hand over to and no retry loop,
+            // so keep the role, just as it was before leader election.
+            if (this.token) {
+                await this.stepDown("startup failed");
+            }
+        }
+    }
+
+    private async stepDown(reason: string): Promise<void> {
+        const wasLeader = this.leader;
+        this.leader = false;
+
+        if (wasLeader) {
+            logger.warn(`Stepping down as leader for ${this.key}: ${reason}`);
+            try {
+                await this.callbacks.onDemoted();
+            } catch (error) {
+                logger.error(
+                    `Error stopping ${this.key} after stepping down:`,
+                    error
+                );
+            }
+        }
+
+        // Only after onDemoted has stopped this instance from scheduling more
+        // work, so the next leader doesn't overlap with it.
+        await this.release();
+    }
+
+    // Best effort: if Redis is unreachable the lease just runs out after
+    // LEASE_TTL_MS instead.
+    private async release(): Promise<void> {
+        const token = this.token;
+        this.token = null;
+
+        const client = redisManager.getWriteClient();
+        if (!token || !client) {
+            return;
+        }
+
+        try {
+            await withTimeout(
+                client.eval(RELEASE_SCRIPT, 1, this.key, token),
+                REDIS_TIMEOUT_MS,
+                `release ${this.key}`
+            );
+        } catch (error) {
+            logger.warn(`Failed to release ${this.key}:`, error);
+        }
+    }
+}

server/routers/auth/deleteMyAccount.ts

@@ -182,7 +182,14 @@ export async function deleteMyAccount(
         const allDeletedNewtIds: string[] = [];
         const allOlmsToTerminate: string[] = [];
 
-        for (const row of ownedOrgsRows) {
+        // Delete billing orgs last: deleting a child org decrements usage on
+        // its billing org, which fails (and aborts the transaction) if the
+        // billing org is already gone.
+        const orgsInDeleteOrder = [...ownedOrgsRows].sort(
+            (a, b) => Number(!!a.isBillingOrg) - Number(!!b.isBillingOrg)
+        );
+
+        for (const row of orgsInDeleteOrder) {
             try {
                 const result = await deleteOrgById(row.orgId);
                 allDeletedNewtIds.push(...result.deletedNewtIds);

server/routers/newt/handleNewtRegisterMessage.ts

@@ -112,7 +112,8 @@ export const handleNewtRegisterMessage: MessageHandler = async (context) => {
             return;
         }
 
-        const newSubnet = await getUniqueSubnetForExitNode(exitNode);
+        const { value: newSubnet, release: releaseSubnetLock } =
+            await getUniqueSubnetForExitNode(exitNode);
 
         if (!newSubnet) {
             logger.error(
@@ -124,15 +125,20 @@ export const handleNewtRegisterMessage: MessageHandler = async (context) => {
 
         siteSubnet = newSubnet;
 
-        await db
-            .update(sites)
-            .set({
-                pubKey: publicKey,
-                exitNodeId: exitNodeId,
-                exitNodeSubnet: newSubnet
-            })
-            .where(eq(sites.siteId, siteId))
-            .returning();
+        try {
+            await db
+                .update(sites)
+                .set({
+                    pubKey: publicKey,
+                    exitNodeId: exitNodeId,
+                    exitNodeSubnet: newSubnet
+                })
+                .where(eq(sites.siteId, siteId))
+                .returning();
+        } finally {
+            // only release once the subnet is persisted so no other node can pick it
+            await releaseSubnetLock();
+        }
     } else {
         await db
             .update(sites)

server/routers/olm/handleOlmRegisterMessage.ts

@@ -313,6 +313,7 @@ export const handleOlmRegisterMessage: MessageHandler = async (context) => {
     }
 
     let clientSubnet = client.exitNodeSubnet;
+    let releaseSubnetLock: (() => Promise<void>) | null = null;
     if (
         exitNodeId &&
         (client.exitNodeId !== exitNodeId || !client.exitNodeSubnet)
@@ -340,7 +341,8 @@ export const handleOlmRegisterMessage: MessageHandler = async (context) => {
 
         // TODO: IF WE DO NOT HAVE AN INFERENCE RESOURCE DO WE NEED TO BE HOLDING A SUBNET ON THE CLIENT?
 
-        const newSubnet = await getUniqueSubnetForExitNode(exitNode);
+        const { value: newSubnet, release } =
+            await getUniqueSubnetForExitNode(exitNode);
 
         if (!newSubnet) {
             logger.error(
@@ -352,6 +354,7 @@ export const handleOlmRegisterMessage: MessageHandler = async (context) => {
         }
 
         clientSubnet = newSubnet;
+        releaseSubnetLock = release;
     }
 
     if (
@@ -364,16 +367,21 @@ export const handleOlmRegisterMessage: MessageHandler = async (context) => {
             "[handleOlmRegisterMessage] Public key mismatch. Updating public key and clearing session info...",
             { orgId: client.orgId, clientId: client.clientId }
         );
-        // Update the client's public key
-        await db
-            .update(clients)
-            .set({
-                pubKey: publicKey,
-                archived: false,
-                exitNodeId: exitNodeId, // this can be undefined if no exit node was selected, which is fine just means we cant talk to the node or connect to it
-                exitNodeSubnet: clientSubnet
-            })
-            .where(eq(clients.clientId, client.clientId));
+        try {
+            // Update the client's public key
+            await db
+                .update(clients)
+                .set({
+                    pubKey: publicKey,
+                    archived: false,
+                    exitNodeId: exitNodeId, // this can be undefined if no exit node was selected, which is fine just means we cant talk to the node or connect to it
+                    exitNodeSubnet: clientSubnet
+                })
+                .where(eq(clients.clientId, client.clientId));
+        } finally {
+            // only release once the subnet is persisted so no other node can pick it
+            await releaseSubnetLock?.();
+        }
 
         // set isRelay to false for all of the client's sites to reset the connection metadata
         await db
@@ -396,6 +404,9 @@ export const handleOlmRegisterMessage: MessageHandler = async (context) => {
             );
     }
 
+    // safety net in case a subnet was locked but the update above was skipped (release is idempotent)
+    await releaseSubnetLock?.();
+
     if (client.pubKey && client.pubKey !== publicKey && client.exitNodeId) {
         // test the old client to see if its different then remove
         logger.info("Public key mismatch. Deleting old peer...");