2ada5acd5a
* refactor(agent): harden local agent volume lifecycle * chore(test): remove un-used variable * refactor(agent): create dedicated jobs for recurring tasks * chore: pr feedbacks * test: add missing fake agent controller
71 lines
2.9 KiB
TypeScript
71 lines
2.9 KiB
TypeScript
import { Job } from "../core/scheduler";
|
|
import path from "node:path";
|
|
import fs from "node:fs/promises";
|
|
import { volumeService } from "../modules/volumes/volume.service";
|
|
import { readMountInfo } from "../utils/mountinfo";
|
|
import { getVolumePath } from "../modules/volumes/helpers";
|
|
import { logger } from "@zerobyte/core/node";
|
|
import { executeUnmount } from "../../../apps/agent/src/volume-host/backends/utils";
|
|
import { toMessage } from "../utils/errors";
|
|
import { VOLUME_MOUNT_BASE } from "../core/constants";
|
|
import { db } from "../db/db";
|
|
import { withContext } from "../core/request-context";
|
|
import { config } from "../core/config";
|
|
import { LOCAL_AGENT_ID } from "../modules/agents/constants";
|
|
|
|
export class CleanupDanglingMountsJob extends Job {
|
|
async run() {
|
|
if (config.flags.enableLocalAgent) {
|
|
logger.debug("Skipping controller-local dangling mount cleanup because local volume execution is agent-owned.");
|
|
return { done: true, timestamp: new Date() };
|
|
}
|
|
|
|
const organizations = await db.query.organization.findMany({});
|
|
if (organizations.length === 0) {
|
|
logger.warn("No organizations found; skipping dangling mount cleanup to avoid false positives.");
|
|
return { done: true, timestamp: new Date() };
|
|
}
|
|
|
|
const allVolumes = [];
|
|
for (const org of organizations) {
|
|
const volumes = await withContext({ organizationId: org.id }, async () => volumeService.listVolumes());
|
|
allVolumes.push(...volumes.filter((volume) => volume.agentId === LOCAL_AGENT_ID));
|
|
}
|
|
|
|
const allSystemMounts = await readMountInfo();
|
|
|
|
for (const mount of allSystemMounts) {
|
|
if (mount.mountPoint.includes("zerobyte") && mount.mountPoint.endsWith("_data")) {
|
|
const matchingVolume = allVolumes.find((v) => getVolumePath(v) === mount.mountPoint);
|
|
if (!matchingVolume) {
|
|
logger.info(`Found dangling mount at ${mount.mountPoint}, attempting to unmount...`);
|
|
await executeUnmount(mount.mountPoint).catch((err) => {
|
|
logger.warn(`Failed to unmount dangling mount ${mount.mountPoint}: ${toMessage(err)}`);
|
|
});
|
|
|
|
await fs.rmdir(path.dirname(mount.mountPoint)).catch((err) => {
|
|
logger.warn(
|
|
`Failed to remove dangling mount directory ${path.dirname(mount.mountPoint)}: ${toMessage(err)}`,
|
|
);
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
const allZerobyteDirs = await fs.readdir(VOLUME_MOUNT_BASE).catch(() => []);
|
|
|
|
for (const dir of allZerobyteDirs) {
|
|
const volumePath = `${VOLUME_MOUNT_BASE}/${dir}/_data`;
|
|
const matchingVolume = allVolumes.find((v) => getVolumePath(v) === volumePath);
|
|
if (!matchingVolume) {
|
|
const fullPath = path.join(VOLUME_MOUNT_BASE, dir);
|
|
logger.info(`Found dangling mount directory at ${fullPath}, attempting to remove...`);
|
|
await fs.rm(fullPath, { recursive: true, force: true }).catch((err) => {
|
|
logger.warn(`Failed to remove dangling mount directory ${fullPath}: ${toMessage(err)}`);
|
|
});
|
|
}
|
|
}
|
|
|
|
return { done: true, timestamp: new Date() };
|
|
}
|
|
}
|