hub: cleanup data of deleted projects

src/packages/backend/files/path-to-files.ts

@@ -0,0 +1,17 @@
+/*
+ *  This file is part of CoCalc: Copyright © 2020 Sagemath, Inc.
+ *  License: AGPLv3 s.t. "Commons Clause" – see LICENSE.md for details
+ */
+
+// This is used to find files on the share server (public_paths) in "next"
+// and also in the hub, for deleting shared files of projects
+
+import { join } from "node:path";
+
+import { projects } from "@cocalc/backend/data";
+
+// Given a project_id/path, return the directory on the file system where
+// that path should be located.
+export function pathToFiles(project_id: string, path: string): string {
+  return join(projects.replace("[project_id]", project_id), path);
+}

src/packages/backend/logger.ts

@@ -12,9 +12,11 @@ process.env.DEBUG_HIDE_DATE = "yes"; // since we supply it ourselves
 // otherwise, maybe stuff like this works: (debug as any).inspectOpts["hideDate"] = true;
 
 import debug, { Debugger } from "debug";
-import { mkdirSync, createWriteStream, statSync, ftruncate } from "fs";
-import { format } from "util";
-import { dirname, join } from "path";
+
+import { createWriteStream, ftruncate, mkdirSync, statSync } from "node:fs";
+import { dirname, join } from "node:path";
+import { format } from "node:util";
+
 import { logs } from "./data";
 
 const MAX_FILE_SIZE_BYTES = 20 * 1024 * 1024; // 20MB
@@ -25,12 +27,12 @@ let _trimLogFileSizePath = "";
 export function trimLogFileSize() {
   // THIS JUST DOESN'T REALLY WORK!
   return;
-  
+
   if (!_trimLogFileSizePath) return;
   let stats;
   try {
     stats = statSync(_trimLogFileSizePath);
-  } catch(_) {
+  } catch (_) {
     // this happens if the file doesn't exist, which is fine since "trimming" it would be a no-op
     return;
   }
@@ -141,27 +143,27 @@ function initTransports() {
 
 initTransports();
 
-const DEBUGGERS = {
-  error: COCALC.extend("error"),
-  warn: COCALC.extend("warn"),
-  info: COCALC.extend("info"),
-  http: COCALC.extend("http"),
-  verbose: COCALC.extend("verbose"),
-  debug: COCALC.extend("debug"),
-  silly: COCALC.extend("silly"),
-};
-
-type Level = keyof typeof DEBUGGERS;
-
-const LEVELS: Level[] = [
+const LEVELS = [
   "error",
   "warn",
   "info",
   "http",
   "verbose",
   "debug",
   "silly",
-];
+] as const;
+
+type Level = (typeof LEVELS)[number];
+
+const DEBUGGERS: { [key in Level]: Debugger } = {
+  error: COCALC.extend("error"),
+  warn: COCALC.extend("warn"),
+  info: COCALC.extend("info"),
+  http: COCALC.extend("http"),
+  verbose: COCALC.extend("verbose"),
+  debug: COCALC.extend("debug"),
+  silly: COCALC.extend("silly"),
+} as const;
 
 class Logger {
   private name: string;
@@ -194,13 +196,13 @@ class Logger {
 }
 
 export interface WinstonLogger {
-  error: Function;
-  warn: Function;
-  info: Function;
-  http: Function;
-  verbose: Function;
-  debug: Function;
-  silly: Function;
+  error: Debugger;
+  warn: Debugger;
+  info: Debugger;
+  http: Debugger;
+  verbose: Debugger;
+  debug: Debugger;
+  silly: Debugger;
   extend: (name: string) => WinstonLogger;
   isEnabled: (level: Level) => boolean;
 }

src/packages/backend/metrics.ts

@@ -1,6 +1,6 @@
 import { Counter, Gauge, Histogram } from "prom-client";
 
-type Aspect = "db" | "database" | "server" | "llm";
+type Aspect = "db" | "database" | "server" | "llm" | "database";
 
 function withPrefix(aspect: Aspect, name: string): string {
   return `cocalc_${aspect}_${name}`;
@@ -13,7 +13,7 @@ export function newCounter(
   name: string,
   help: string,
   labelNames: string[] = [],
-) {
+): Counter<string> {
   name = withPrefix(aspect, name);
   const key = `counter-${name}`;
   if (cache[key] != null) {

src/packages/backend/misc.ts

@@ -1,4 +1,7 @@
 import { createHash } from "crypto";
+import { join } from "node:path";
+
+import { projects } from "@cocalc/backend/data";
 import { is_valid_uuid_string } from "@cocalc/util/misc";
 
 /*
@@ -69,3 +72,14 @@ export function envForSpawn() {
   }
   return env;
 }
+
+// return the absolute home directory of given @project_id project on disk
+export function homePath(project_id: string): string {
+  // $MOUNTED_PROJECTS_ROOT is for OnPrem and that "projects" location is only for dev/single-user
+  const projects_root = process.env.MOUNTED_PROJECTS_ROOT;
+  if (projects_root) {
+    return join(projects_root, project_id);
+  } else {
+    return projects.replace("[project_id]", project_id);
+  }
+}

src/packages/database/jest.config.js

@@ -3,4 +3,5 @@ module.exports = {
   preset: 'ts-jest',
   testEnvironment: 'node',
   testMatch: ['**/?(*.)+(spec|test).ts?(x)'],
+  setupFiles: ['./test/setup.js'], // Path to your setup file
 };

src/packages/database/package.json

@@ -23,7 +23,7 @@
     "@cocalc/util": "workspace:*",
     "@qdrant/js-client-rest": "^1.6.0",
     "@types/lodash": "^4.14.202",
-    "@types/pg": "^8.6.1",
+    "@types/pg": "^8.11.6",
     "@types/uuid": "^8.3.1",
     "async": "^1.5.2",
     "awaiting": "^3.0.0",
@@ -34,7 +34,7 @@
     "lodash": "^4.17.21",
     "lru-cache": "^7.14.1",
     "node-fetch": "2.6.7",
-    "pg": "^8.7.1",
+    "pg": "^8.12.0",
     "random-key": "^0.3.2",
     "read": "^1.0.7",
     "sql-string-escape": "^1.1.6",
@@ -50,7 +50,7 @@
     "build": "../node_modules/.bin/tsc  --build && coffee -c -o dist/ ./",
     "clean": "rm -rf dist",
     "tsc": "../node_modules/.bin/tsc --watch  --pretty --preserveWatchOutput",
-    "test": "pnpm exec jest",
+    "test": "TZ=UTC jest --forceExit --runInBand",
     "prepublishOnly": "pnpm test"
   },
   "repository": {

src/packages/database/postgres-server-queries.coffee

@@ -51,7 +51,7 @@ read = require('read')
 {site_license_manager_set} = require('./postgres/site-license/manager')
 {matching_site_licenses, manager_site_licenses} = require('./postgres/site-license/search')
 {project_datastore_set, project_datastore_get, project_datastore_del} = require('./postgres/project-queries')
-{permanently_unlink_all_deleted_projects_of_user, unlink_old_deleted_projects} = require('./postgres/delete-projects')
+{permanently_unlink_all_deleted_projects_of_user, unlink_old_deleted_projects, cleanup_old_projects_data} = require('./postgres/delete-projects')
 {get_all_public_paths, unlist_all_public_paths} = require('./postgres/public-paths')
 {get_personal_user} = require('./postgres/personal')
 {set_passport_settings, get_passport_settings, get_all_passport_settings, get_all_passport_settings_cached, create_passport, passport_exists, update_account_and_passport, _passport_key} = require('./postgres/passport')
@@ -2592,6 +2592,10 @@ exports.extend_PostgreSQL = (ext) -> class PostgreSQL extends ext
     unlink_old_deleted_projects: () =>
         return await unlink_old_deleted_projects(@)
 
+    # async function
+    cleanup_old_projects_data: (max_run_m) =>
+        return await cleanup_old_projects_data(@, max_run_m)
+
     # async function
     unlist_all_public_paths: (account_id, is_owner) =>
         return await unlist_all_public_paths(@, account_id, is_owner)

src/packages/database/postgres/bulk-delete.test.ts

@@ -0,0 +1,70 @@
+/*
+ *  This file is part of CoCalc: Copyright © 2024 Sagemath, Inc.
+ *  License: AGPLv3 s.t. "Commons Clause" – see LICENSE.md for details
+ */
+
+import getPool, { initEphemeralDatabase } from "@cocalc/database/pool";
+import { uuid } from "@cocalc/util/misc";
+import { bulkDelete } from "./bulk-delete";
+
+beforeAll(async () => {
+  await initEphemeralDatabase({});
+}, 15000);
+
+afterAll(async () => {
+  await getPool().end();
+});
+
+describe("bulk delete", () => {
+  test("deleting projects", async () => {
+    const p = getPool();
+    const project_id = uuid();
+    const N = 100000;
+
+    // extra entry, which has to remain
+    const other = uuid();
+    await p.query(
+      "INSERT INTO project_log (id, project_id, time) VALUES($1::UUID, $2::UUID, $3::TIMESTAMP)",
+      [other, uuid(), new Date()],
+    );
+
+    await p.query(
+      `INSERT INTO project_log (id, project_id, time)
+      SELECT gen_random_uuid(), $1::UUID, NOW() - interval '1 second' * g.n
+      FROM generate_series(1, $2) AS g(n)`,
+      [project_id, N],
+    );
+
+    const num1 = await p.query(
+      "SELECT COUNT(*)::INT as num FROM project_log WHERE project_id = $1",
+      [project_id],
+    );
+    expect(num1.rows[0].num).toEqual(N);
+
+    const res = await bulkDelete({
+      table: "project_log",
+      field: "project_id",
+      value: project_id,
+    });
+
+    // if this ever fails, the "ret.rowCount" value is inaccurate.
+    // This must be replaced by "RETURNING 1" in the the query and a "SELECT COUNT(*) ..." and so.
+    // (and not only here, but everywhere in the code base)
+    expect(res.rowsDeleted).toEqual(N);
+    expect(res.durationS).toBeGreaterThan(0.1);
+    expect(res.totalPgTimeS).toBeGreaterThan(0.1);
+    expect(res.totalWaitS).toBeGreaterThan(0.1);
+    expect((res.totalPgTimeS * 10) / res.totalWaitS).toBeGreaterThan(0.5);
+
+    const num2 = await p.query(
+      "SELECT COUNT(*)::INT as num FROM project_log WHERE project_id = $1",
+      [project_id],
+    );
+    expect(num2.rows[0].num).toEqual(0);
+
+    const otherRes = await p.query("SELECT * FROM project_log WHERE id = $1", [
+      other,
+    ]);
+    expect(otherRes.rows[0].id).toEqual(other);
+  }, 10000);
+});

src/packages/database/postgres/bulk-delete.ts

@@ -0,0 +1,98 @@
+import { escapeIdentifier } from "pg";
+
+import getLogger from "@cocalc/backend/logger";
+import { envToInt } from "@cocalc/backend/misc/env-to-number";
+import getPool from "@cocalc/database/pool";
+import { SCHEMA } from "@cocalc/util/schema";
+
+const log = getLogger("db:bulk-delete");
+const D = log.debug;
+
+type Field =
+  | "project_id"
+  | "account_id"
+  | "target_project_id"
+  | "source_project_id";
+
+const MAX_UTIL_PCT = envToInt("COCALC_DB_BULK_DELETE_MAX_UTIL_PCT", 10);
+// adjust the time limits: by default, we aim to keep the operation between 0.1 and 0.2 secs
+const MAX_TIME_TARGET_MS = envToInt(
+  "COCALC_DB_BULK_DELETE_MAX_TIME_TARGET_MS",
+  100,
+);
+const MAX_TARGET_S = MAX_TIME_TARGET_MS / 1000;
+const MIN_TARGET_S = MAX_TARGET_S / 2;
+const DEFAULT_LIMIT = envToInt("COCALC_DB_BULK_DELETE_DEFAULT_LIMIT", 16);
+const MAX_LIMIT = envToInt("COCALC_DB_BULK_DELETE_MAX_LIMIT", 32768);
+
+interface Opts {
+  table: string; // e.g. project_log, etc.
+  field: Field; // for now, we only support a few
+  id?: string; // default "id", the ID field in the table, which identifies each row uniquely
+  value: string; // a UUID
+  limit?: number; // default 1024
+  maxUtilPct?: number; // 0-100, percent
+}
+
+type Ret = Promise<{
+  rowsDeleted: number;
+  durationS: number;
+  totalWaitS: number;
+  totalPgTimeS: number;
+}>;
+
+function deleteQuery(table: string, field: string, id: string) {
+  const T = escapeIdentifier(table);
+  const F = escapeIdentifier(field);
+  const ID = escapeIdentifier(id);
+
+  return `
+DELETE FROM ${T}
+WHERE ${ID} IN (
+    SELECT ${ID} FROM ${T} WHERE ${F} = $1 LIMIT $2
+)`;
+}
+
+export async function bulkDelete(opts: Opts): Ret {
+  const { table, field, value, id = "id", maxUtilPct = MAX_UTIL_PCT } = opts;
+  let { limit = DEFAULT_LIMIT } = opts;
+  // assert table name is a key in SCHEMA
+  if (!(table in SCHEMA)) {
+    throw new Error(`table ${table} does not exist`);
+  }
+
+  if (maxUtilPct < 1 || maxUtilPct > 99) {
+    throw new Error(`maxUtilPct must be between 1 and 99`);
+  }
+
+  const q = deleteQuery(table, field, id);
+  const pool = getPool();
+  const start_ts = Date.now();
+
+  let rowsDeleted = 0;
+  let totalWaitS = 0;
+  let totalPgTimeS = 0;
+  while (true) {
+    const t0 = Date.now();
+    const ret = await pool.query(q, [value, limit]);
+    const dt = (Date.now() - t0) / 1000;
+    rowsDeleted += ret.rowCount ?? 0;
+    totalPgTimeS += dt;
+
+    const next =
+      dt > MAX_TARGET_S ? limit / 2 : dt < MIN_TARGET_S ? limit * 2 : limit;
+    limit = Math.max(1, Math.min(MAX_LIMIT, Math.round(next)));
+
+    // wait for a bit, but not more than 1 second ~ this aims for a max utilization of 10%
+    const waitS = Math.min(1, dt * ((100 - maxUtilPct) / maxUtilPct));
+    await new Promise((done) => setTimeout(done, 1000 * waitS));
+    totalWaitS += waitS;
+
+    D(`deleted ${ret.rowCount} | dt=${dt} | wait=${waitS} | limit=${limit}`);
+
+    if (ret.rowCount === 0) break;
+  }
+
+  const durationS = (Date.now() - start_ts) / 1000;
+  return { durationS, rowsDeleted, totalWaitS, totalPgTimeS };
+}