remove controller service lock

app/src/adapters/database/repositories/queue.ts

@@ -9,40 +9,6 @@ import {
 export class QueueRepository {
     constructor(private readonly context: DatabaseContext) {}
 
-    acquireLock(name: string, owner: string, ttlMs: number): boolean {
-        return this.context.transaction(() => {
-            const time = now();
-            this.context.db
-                .prepare(
-                    "DELETE FROM service_locks WHERE name = $name AND expires_at < $now",
-                )
-                .run({ $name: name, $now: time });
-            const result = this.context.db
-                .prepare(
-                    "INSERT OR IGNORE INTO service_locks(name, owner, expires_at) VALUES ($name, $owner, $expires)",
-                )
-                .run({ $name: name, $owner: owner, $expires: time + ttlMs });
-            return Number(result.changes) === 1;
-        });
-    }
-
-    renewLock(name: string, owner: string, ttlMs: number): boolean {
-        const result = this.context.db
-            .prepare(
-                "UPDATE service_locks SET expires_at = $expires WHERE name = $name AND owner = $owner",
-            )
-            .run({ $name: name, $owner: owner, $expires: now() + ttlMs });
-        return Number(result.changes) === 1;
-    }
-
-    releaseLock(name: string, owner: string): void {
-        this.context.db
-            .prepare(
-                "DELETE FROM service_locks WHERE name = $name AND owner = $owner",
-            )
-            .run({ $name: name, $owner: owner });
-    }
-
     recover(): void {
         this.context.db.exec(`
       UPDATE webhook_deliveries SET status = 'pending' WHERE status = 'processing';

app/src/adapters/database/schema.ts

@@ -116,11 +116,6 @@ export function migrate(db: DatabaseSync): void {
       created_at INTEGER NOT NULL,
       FOREIGN KEY(job_id) REFERENCES jobs(id) ON DELETE SET NULL
     ) STRICT;
-    CREATE TABLE IF NOT EXISTS service_locks (
-      name TEXT PRIMARY KEY,
-      owner TEXT NOT NULL,
-      expires_at INTEGER NOT NULL
-    ) STRICT;
     INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES (1, unixepoch('subsec') * 1000);
   `);
 }

app/src/adapters/database/store.ts

@@ -84,15 +84,6 @@ export class AgentStore implements DatabaseContext {
             });
     }
 
-    acquireServiceLock(name: string, owner: string, ttlMs: number): boolean {
-        return this.queue.acquireLock(name, owner, ttlMs);
-    }
-    renewServiceLock(name: string, owner: string, ttlMs: number): boolean {
-        return this.queue.renewLock(name, owner, ttlMs);
-    }
-    releaseServiceLock(name: string, owner: string): void {
-        this.queue.releaseLock(name, owner);
-    }
     recoverControllerWork(): void {
         this.queue.recover();
     }

app/src/application/controller/main.ts

@@ -1,5 +1,4 @@
 #!/usr/bin/env node
-import { randomUUID } from "node:crypto";
 import { createServer } from "node:http";
 import { pathToFileURL } from "node:url";
 import { AgentStore } from "../../adapters/database/store.js";
@@ -25,9 +24,6 @@ async function main(): Promise<void> {
     const store = new AgentStore(
         process.env.AGENT_DB_PATH || "/var/lib/gitea-agent/agent.db",
     );
-    const owner = `controller-${randomUUID()}`;
-    if (!store.acquireServiceLock("controller", owner, 30_000))
-        throw new Error("Another controller owns the service lock");
     store.recoverControllerWork();
     store.purgeDeliveries(Date.now() - 7 * 24 * 60 * 60_000);
     const shutdown = new AbortController();
@@ -93,13 +89,11 @@ async function main(): Promise<void> {
         }
     };
     const pumpTimer = setInterval(pump, 250);
-    const lockTimer = setInterval(() => renewLock(store, owner), 5_000);
     const retentionTimer = setInterval(
         () => store.purgeDeliveries(Date.now() - 7 * 24 * 60 * 60_000),
         60 * 60_000,
     );
     pumpTimer.unref();
-    lockTimer.unref();
     retentionTimer.unref();
 
     const server = createServer((request, response) => {
@@ -141,7 +135,6 @@ async function main(): Promise<void> {
             if (stopping) return;
             stopping = true;
             clearInterval(pumpTimer);
-            clearInterval(lockTimer);
             clearInterval(retentionTimer);
             server.close(() => resolve());
             shutdown.abort(new Error("Controller is shutting down"));
@@ -153,24 +146,9 @@ async function main(): Promise<void> {
         delivering?.catch(() => undefined),
         publishing?.catch(() => undefined),
     ]);
-    store.releaseServiceLock("controller", owner);
     store.close();
 }
 
-function renewLock(store: AgentStore, owner: string): void {
-    try {
-        if (!store.renewServiceLock("controller", owner, 30_000))
-            process.kill(process.pid, "SIGTERM");
-    } catch (error) {
-        console.error(
-            log("error", "Controller lock renewal failed", {
-                error: safeFailure(error),
-            }),
-        );
-        process.kill(process.pid, "SIGTERM");
-    }
-}
-
 function parsePort(value: string): number {
     const port = Number(value);
     if (!Number.isInteger(port) || port < 1 || port > 65_535)

app/src/tests/adapters/store.test.ts

@@ -202,32 +202,6 @@ test("a replayed cancel command remains bound to its original job", async () =>
     }
 });
 
-test("the durable controller lock prevents overlapping publishers", async () => {
-    const root = await mkdtemp(join(tmpdir(), "agent-lock-"));
-    const path = join(root, "agent.db");
-    const first = new AgentStore(path);
-    const second = new AgentStore(path);
-    try {
-        assert.equal(
-            first.acquireServiceLock("controller", "one", 30_000),
-            true,
-        );
-        assert.equal(
-            second.acquireServiceLock("controller", "two", 30_000),
-            false,
-        );
-        first.releaseServiceLock("controller", "one");
-        assert.equal(
-            second.acquireServiceLock("controller", "two", 30_000),
-            true,
-        );
-    } finally {
-        second.close();
-        first.close();
-        await rm(root, { recursive: true, force: true });
-    }
-});
-
 function required<T>(value: T | undefined): T {
     assert.ok(value);
     return value;