Persist local actor signing keys

packages/drfed/package.json

@@ -53,6 +53,10 @@
     "./serving": {
       "types": "./dist/serving.d.mts",
       "default": "./dist/serving.mjs"
+    },
+    "./query-logger": {
+      "types": "./dist/query-logger.d.mts",
+      "default": "./dist/query-logger.mjs"
     }
   },
   "files": [
@@ -67,7 +71,8 @@
     "entry": [
       "src/index.ts",
       "src/valueparser.ts",
-      "src/serving.ts"
+      "src/serving.ts",
+      "src/query-logger.ts"
     ],
     "dts": {
       "sourcemap": true,

packages/drfed/src/index.ts

@@ -18,12 +18,13 @@ import { writeFile } from "node:fs/promises";
 import process from "node:process";
 
 import createFederation, { createInboundRecorder } from "@drfed/federation";
+import { KeyGenerationQueue } from "@drfed/federation/task-queue";
 import { createYogaServer } from "@drfed/graphql";
 import { schema } from "@drfed/graphql/schema";
 import { migrate } from "@drfed/models";
 import { PgliteKvStore } from "@fedify/pglite";
 import { PostgresKvStore } from "@fedify/postgres";
-import { configure, getConsoleSink } from "@logtape/logtape";
+import { configure, getConsoleSink, getLogger } from "@logtape/logtape";
 import { createLoggingConfig } from "@optique/logtape";
 import { run } from "@optique/run";
 import { SmtpTransport } from "@upyo/smtp";
@@ -50,8 +51,21 @@ async function runServer(options: ServerOptions) {
       : new PostgresKvStore(credentials.client);
   const federation = await createFederation(options.drizzle.db, {
     kv,
+    queue: { task: new KeyGenerationQueue() },
+    taskQueueResolution: "strict",
+    manuallyStartQueue: true,
     allowPrivateAddress: true,
   });
+  const workerAbort = new AbortController();
+  // oxlint-disable promise/prefer-await-to-then
+  const worker = federation
+    .startQueue(undefined, { queue: "task", signal: workerAbort.signal })
+    .catch(() => {
+      getLogger(["drfed", "server"]).error(
+        "Actor key worker stopped unexpectedly.",
+      );
+    });
+  // oxlint-enable promise/prefer-await-to-then
   const { emailFrom, mailer, rootOrigin, loginOrigins } = options;
 
   const yogaServer = createYogaServer(options.drizzle.db, federation, {
@@ -73,23 +87,55 @@ async function runServer(options: ServerOptions) {
     }),
     hostname: options.address.host,
     manual: true,
+    gracefulShutdown: false,
     port: options.address.port,
   });
+  let closing = false;
   function shutdown() {
-    if (mailer instanceof SmtpTransport) {
-      mailer.closeAllConnections();
+    if (closing) {
+      process.exit(1);
     }
-    // oxlint-disable-next-line promise/catch-or-return promise/prefer-await-to-then
-    server.close().then(async () => {
-      await ("driver" in credentials
-        ? credentials.client.close()
-        : credentials.client.end());
-      process.exit(0);
-    });
+    closing = true;
+    const deadline = setTimeout(() => process.exit(1), 10_000);
+    const requests = server.close();
+    const forceClose = setTimeout(() => {
+      // A stalled upload must not prevent database cleanup on shutdown.
+      // oxlint-disable-next-line promise/prefer-await-to-then
+      void server.close(true).catch(() => process.exit(1));
+    }, 5000);
+    workerAbort.abort();
+    // The task worker awaits its active handler before resolving.
+    // oxlint-disable promise/prefer-await-to-then
+    Promise.all([requests, worker])
+      .then(async () => {
+        clearTimeout(forceClose);
+        if (mailer instanceof SmtpTransport) mailer.closeAllConnections();
+        await ("driver" in credentials
+          ? credentials.client.close()
+          : credentials.client.end());
+        clearTimeout(deadline);
+        process.exit(0);
+      })
+      .catch(() => {
+        process.exit(1);
+      });
+  }
+  // oxlint-enable promise/prefer-await-to-then
+  process.on("SIGINT", shutdown);
+  process.on("SIGTERM", shutdown);
+  try {
+    await server.serve();
+  } catch (error) {
+    process.off("SIGINT", shutdown);
+    process.off("SIGTERM", shutdown);
+    workerAbort.abort();
+    await Promise.all([server.close(), worker]);
+    if (mailer instanceof SmtpTransport) mailer.closeAllConnections();
+    await ("driver" in credentials
+      ? credentials.client.close()
+      : credentials.client.end());
+    throw new Error("Could not start the server.", { cause: error });
   }
-  process.once("SIGINT", shutdown);
-  process.once("SIGTERM", shutdown);
-  await server.serve();
 }
 
 async function runSchemaGenerator(

packages/drfed/src/lifecycle.test.ts

@@ -0,0 +1,151 @@
+// DrFed: A web-based platform for developing and debugging ActivityPub apps
+// Copyright (C) 2026 DrFed team
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { once } from "node:events";
+import { mkdtemp, rm } from "node:fs/promises";
+import { type Server, connect, createServer } from "node:net";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import process from "node:process";
+import { it } from "node:test";
+import { setTimeout as delay } from "node:timers/promises";
+import { fileURLToPath } from "node:url";
+
+const binary = fileURLToPath(
+  new URL("../bin/drfed-server.mjs", import.meta.resolve("@drfed/drfed")),
+);
+async function reservePort() {
+  const server = createServer();
+  server.listen(0, "127.0.0.1");
+  await once(server, "listening");
+  const address = server.address();
+  assert.ok(address != null && typeof address !== "string");
+  return { server, port: address.port };
+}
+async function closeServer(server: Server) {
+  const closed = once(server, "close");
+  server.close();
+  await closed;
+}
+function startServer(port: number, dataPath: string) {
+  const child = spawn(
+    process.execPath,
+    [
+      binary,
+      "--root-origin=http://drfed.test",
+      "--login-origin=http://drfed.test",
+      `--listen=127.0.0.1:${port}`,
+      `--pglite-data-path=${dataPath}`,
+      "--log-level=error",
+    ],
+    { stdio: ["ignore", "pipe", "pipe"] },
+  );
+  let stderr = "";
+  child.stderr.on("data", (chunk) => {
+    stderr += chunk.toString();
+  });
+  child.stdout.resume();
+  const exited = once(child, "close");
+  return { child, exited, stderr: () => stderr };
+}
+
+async function waitForExit(run: ReturnType<typeof startServer>) {
+  const stopTimeout = new AbortController();
+  const timeout = async () => {
+    await delay(90_000, undefined, { signal: stopTimeout.signal });
+    assert.fail(`CLI did not exit within 90 seconds: ${run.stderr()}`);
+  };
+  try {
+    return await Promise.race([run.exited, timeout()]);
+  } finally {
+    stopTimeout.abort();
+  }
+}
+
+it("preserves a listen error when server startup fails", async () => {
+  const { server, port } = await reservePort();
+  const dataPath = await mkdtemp(join(tmpdir(), "drfed-startup-"));
+  const run = startServer(port, dataPath);
+  try {
+    const [code] = await waitForExit(run);
+    assert.equal(code, 1);
+    assert.match(run.stderr(), /EADDRINUSE/u);
+  } finally {
+    run.child.kill("SIGKILL");
+    await closeServer(server);
+    await rm(dataPath, { recursive: true, force: true });
+  }
+});
+
+it(
+  "force-closes a stalled upload and exits normally on SIGTERM",
+  {
+    // Windows child.kill("SIGTERM") forcibly terminates without running handlers.
+    skip: process.platform === "win32",
+  },
+  async () => {
+    const { server, port } = await reservePort();
+    await closeServer(server);
+    const dataPath = await mkdtemp(join(tmpdir(), "drfed-shutdown-"));
+    const run = startServer(port, dataPath);
+    try {
+      let ready = false;
+      const readinessDeadline = performance.now() + 90_000;
+      while (performance.now() < readinessDeadline) {
+        assert.equal(run.child.exitCode, null, run.stderr());
+        try {
+          // oxlint-disable-next-line no-await-in-loop
+          const response = await fetch(`http://127.0.0.1:${port}/graphql`, {
+            method: "POST",
+            signal: AbortSignal.timeout(1000),
+            headers: { "content-type": "application/json" },
+            body: '{"query":"{ __typename }"}',
+          });
+          if (response.ok) {
+            ready = true;
+            break;
+          }
+        } catch {
+          // Migrations and listening have not finished yet.
+        }
+        // oxlint-disable-next-line no-await-in-loop
+        await delay(100);
+      }
+      assert.ok(ready, run.stderr());
+      const stalled = connect({ host: "127.0.0.1", port });
+      // Force-closing a stalled upload may reset its socket.
+      stalled.on("error", () => undefined);
+      try {
+        await once(stalled, "connect");
+        stalled.write(
+          "POST /graphql HTTP/1.1\r\nHost: drfed.test\r\nContent-Type: application/json\r\nContent-Length: 100\r\n\r\n{",
+        );
+        await delay(100);
+        run.child.kill("SIGTERM");
+        const [code, signal] = await waitForExit(run);
+        assert.equal(signal, null);
+        assert.equal(code, 0, run.stderr());
+      } finally {
+        stalled.destroy();
+      }
+    } finally {
+      run.child.kill("SIGKILL");
+      await rm(dataPath, { recursive: true, force: true });
+    }
+  },
+);

packages/drfed/src/parser.ts

@@ -16,7 +16,6 @@
 
 import { relations, schema } from "@drfed/models";
 import { PGlite } from "@electric-sql/pglite";
-import { getLogger } from "@logtape/drizzle-orm";
 import { merge, object, or } from "@optique/core/constructs";
 import { message, optionNames } from "@optique/core/message";
 import { map, multiple, optional, withDefault } from "@optique/core/modifiers";
@@ -31,6 +30,7 @@ import { drizzle as drizzlePglite } from "drizzle-orm/pglite";
 import { drizzle as drizzlePostgres } from "drizzle-orm/postgres-js";
 import postgres from "postgres";
 
+import { privateKeySafeLogger } from "./query-logger.ts";
 import { rootOrigin } from "./valueparser.ts";
 
 const pgliteParser = map(
@@ -54,7 +54,7 @@ const pgliteParser = map(
         client,
         relations,
         schema,
-        logger: getLogger(),
+        logger: privateKeySafeLogger(),
       }),
     };
   },
@@ -80,7 +80,7 @@ const postgresParser = map(
         client,
         relations,
         schema,
-        logger: getLogger(),
+        logger: privateKeySafeLogger(),
       }),
     };
   },

packages/drfed/src/query-logger.test.ts

@@ -0,0 +1,36 @@
+// DrFed: A web-based platform for developing and debugging ActivityPub apps
+// Copyright (C) 2026 DrFed team
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import assert from "node:assert/strict";
+import { it } from "node:test";
+
+import { privateKeySafeLogger } from "@drfed/drfed/query-logger";
+
+it("never forwards key parameters to the ordinary SQL logger", () => {
+  const calls: unknown[] = [];
+  const logger = privateKeySafeLogger({
+    logQuery(query, params) {
+      calls.push([query, params]);
+    },
+  });
+  logger.logQuery('insert into "local_actor_keys" values ($1)', [
+    "PRIVATE_SECRET",
+  ]);
+  logger.logQuery('select * from "local_actor_keys"', []);
+  assert.deepEqual(calls, []);
+  logger.logQuery("select $1", [42]);
+  assert.deepEqual(calls, [["select $1", [42]]]);
+});

packages/drfed/src/query-logger.ts

@@ -0,0 +1,34 @@
+// DrFed: A web-based platform for developing and debugging ActivityPub apps
+// Copyright (C) 2026 DrFed team
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import { getLogger as getQueryLogger } from "@logtape/drizzle-orm";
+import { getLogger } from "@logtape/logtape";
+import type { Logger } from "drizzle-orm/logger";
+/** Suppress parameters for SQL that reads or writes private signing material.
+ * @returns A logger that never logs private-key parameters.
+ */
+export function privateKeySafeLogger(
+  delegate: Logger = getQueryLogger(),
+): Logger {
+  const logger = getLogger(["drfed", "database"]);
+  return {
+    logQuery(query, params) {
+      if (query.includes("local_actor_keys")) {
+        logger.debug("Query: {query}", { query });
+      } else delegate.logQuery(query, params);
+    },
+  };
+}