mirror of
https://github.com/luckyyzh/pi-agent-integrated.git
synced 2026-10-03 02:59:35 +00:00
add automatic storage maintenance (#4)
This commit is contained in:
Zihao Yang
+16
-1
@@ -1,6 +1,11 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { join } from "node:path";
|
||||
import { managedEnvironment, rootDir } from "./profile.mjs";
|
||||
import {
|
||||
isLocalPortListening,
|
||||
maintainStorage,
|
||||
printMaintenanceResult,
|
||||
} from "./storage-maintenance.mjs";
|
||||
|
||||
const target = process.argv[2] ?? "dev";
|
||||
const allowedTargets = new Set(["build", "dev", "dev:lan", "start", "start:lan"]);
|
||||
@@ -15,12 +20,19 @@ if (!npmCliPath) {
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const childEnvironment = managedEnvironment();
|
||||
if (await isLocalPortListening(30141)) {
|
||||
console.log("[storage] skipped automatic maintenance because Pi Web is already running");
|
||||
} else {
|
||||
printMaintenanceResult(maintainStorage({ mode: "auto", env: childEnvironment }));
|
||||
}
|
||||
|
||||
const child = spawn(
|
||||
process.execPath,
|
||||
[npmCliPath, "--prefix", join(rootDir, "pi-web"), "run", target],
|
||||
{
|
||||
cwd: rootDir,
|
||||
env: managedEnvironment(),
|
||||
env: childEnvironment,
|
||||
stdio: "inherit",
|
||||
},
|
||||
);
|
||||
@@ -34,6 +46,9 @@ child.on("error", (error) => {
|
||||
process.exit(1);
|
||||
});
|
||||
child.on("exit", (code, signal) => {
|
||||
if (!signal && code === 0 && target === "build") {
|
||||
printMaintenanceResult(maintainStorage({ mode: "auto", env: childEnvironment }));
|
||||
}
|
||||
if (signal) process.kill(process.pid, signal);
|
||||
process.exit(code ?? 1);
|
||||
});
|
||||
|
||||
@@ -0,0 +1,255 @@
|
||||
import { spawnSync } from "node:child_process";
|
||||
import {
|
||||
existsSync,
|
||||
lstatSync,
|
||||
readdirSync,
|
||||
rmSync,
|
||||
} from "node:fs";
|
||||
import { createConnection } from "node:net";
|
||||
import { basename, join, resolve } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { dataDir as managedDataDir, rootDir as projectRootDir } from "./profile.mjs";
|
||||
|
||||
const MIB = 1024 * 1024;
|
||||
const DAY_MS = 24 * 60 * 60 * 1000;
|
||||
|
||||
function numericSetting(env, name, fallback) {
|
||||
const value = Number.parseFloat(env[name] ?? "");
|
||||
return Number.isFinite(value) && value >= 0 ? value : fallback;
|
||||
}
|
||||
|
||||
function enabledSetting(env, name, fallback = true) {
|
||||
const value = env[name]?.trim().toLowerCase();
|
||||
if (!value) return fallback;
|
||||
return !["0", "false", "no", "off"].includes(value);
|
||||
}
|
||||
|
||||
export function storagePolicy(env = process.env) {
|
||||
return {
|
||||
enabled: enabledSetting(env, "PI_STORAGE_AUTO_MAINTENANCE", true),
|
||||
nextDevMaxBytes: numericSetting(env, "PI_STORAGE_NEXT_DEV_MAX_MB", 640) * MIB,
|
||||
nextBuildCacheMaxBytes: numericSetting(env, "PI_STORAGE_NEXT_CACHE_MAX_MB", 256) * MIB,
|
||||
npmCacheMaxBytes: numericSetting(env, "PI_STORAGE_NPM_CACHE_MAX_MB", 256) * MIB,
|
||||
orphanGraceMs: numericSetting(env, "PI_STORAGE_ORPHAN_GRACE_DAYS", 7) * DAY_MS,
|
||||
rewindGcLooseObjects: numericSetting(env, "PI_STORAGE_REWIND_GC_OBJECTS", 1000),
|
||||
};
|
||||
}
|
||||
|
||||
export function directoryStats(path) {
|
||||
if (!existsSync(path)) return { bytes: 0, files: 0, newestMtimeMs: 0 };
|
||||
|
||||
let bytes = 0;
|
||||
let files = 0;
|
||||
let newestMtimeMs = 0;
|
||||
const pending = [path];
|
||||
|
||||
while (pending.length > 0) {
|
||||
const current = pending.pop();
|
||||
let entries;
|
||||
try {
|
||||
entries = readdirSync(current, { withFileTypes: true });
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const entry of entries) {
|
||||
const child = join(current, entry.name);
|
||||
let stats;
|
||||
try {
|
||||
stats = lstatSync(child);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
newestMtimeMs = Math.max(newestMtimeMs, stats.mtimeMs);
|
||||
if (stats.isSymbolicLink()) continue;
|
||||
if (stats.isDirectory()) pending.push(child);
|
||||
else if (stats.isFile()) {
|
||||
bytes += stats.size;
|
||||
files += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return { bytes, files, newestMtimeMs };
|
||||
}
|
||||
|
||||
function collectSessionIds(path) {
|
||||
const ids = new Set();
|
||||
if (!existsSync(path)) return ids;
|
||||
const pending = [path];
|
||||
while (pending.length > 0) {
|
||||
const current = pending.pop();
|
||||
let entries;
|
||||
try {
|
||||
entries = readdirSync(current, { withFileTypes: true });
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
for (const entry of entries) {
|
||||
const child = join(current, entry.name);
|
||||
if (entry.isDirectory()) pending.push(child);
|
||||
else if (entry.isFile() && entry.name.endsWith(".jsonl")) ids.add(basename(entry.name, ".jsonl"));
|
||||
}
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
|
||||
function countLooseGitObjects(checkpointPath) {
|
||||
const objectsPath = join(checkpointPath, ".git", "objects");
|
||||
if (!existsSync(objectsPath)) return 0;
|
||||
let count = 0;
|
||||
for (const entry of readdirSync(objectsPath, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory() || !/^[0-9a-f]{2}$/i.test(entry.name)) continue;
|
||||
count += readdirSync(join(objectsPath, entry.name), { withFileTypes: true })
|
||||
.filter((child) => child.isFile()).length;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
function defaultCheckpointCompactor(path) {
|
||||
const result = spawnSync("git", ["-C", path, "gc", "--quiet"], {
|
||||
encoding: "utf8",
|
||||
windowsHide: true,
|
||||
});
|
||||
if (result.error) throw result.error;
|
||||
if (result.status !== 0) throw new Error(result.stderr.trim() || `git gc exited with ${result.status}`);
|
||||
}
|
||||
|
||||
function removeDirectory(path, label, result) {
|
||||
const before = directoryStats(path);
|
||||
if (!existsSync(path)) return;
|
||||
rmSync(path, { recursive: true, force: true });
|
||||
result.reclaimedBytes += before.bytes;
|
||||
result.actions.push({ type: "removed", label, path, bytes: before.bytes });
|
||||
}
|
||||
|
||||
export function maintainStorage({
|
||||
mode = "auto",
|
||||
rootDir = projectRootDir,
|
||||
dataDir = managedDataDir,
|
||||
env = process.env,
|
||||
nowMs = Date.now(),
|
||||
compactCheckpoint = defaultCheckpointCompactor,
|
||||
} = {}) {
|
||||
const policy = storagePolicy(env);
|
||||
const result = { mode, actions: [], warnings: [], reclaimedBytes: 0, status: {} };
|
||||
if (mode === "auto" && !policy.enabled) return { ...result, disabled: true };
|
||||
|
||||
const cacheTargets = [
|
||||
{
|
||||
label: "Next.js development output",
|
||||
path: join(rootDir, "pi-web", ".next", "dev"),
|
||||
limit: policy.nextDevMaxBytes,
|
||||
},
|
||||
{
|
||||
label: "Next.js build cache",
|
||||
path: join(rootDir, "pi-web", ".next", "cache"),
|
||||
limit: policy.nextBuildCacheMaxBytes,
|
||||
},
|
||||
{
|
||||
label: "managed npm cache",
|
||||
path: join(dataDir, "cache", "npm"),
|
||||
limit: policy.npmCacheMaxBytes,
|
||||
},
|
||||
];
|
||||
|
||||
for (const target of cacheTargets) {
|
||||
const stats = directoryStats(target.path);
|
||||
result.status[target.label] = stats;
|
||||
if (mode === "clean" || (mode === "auto" && stats.bytes > target.limit)) {
|
||||
removeDirectory(target.path, target.label, result);
|
||||
}
|
||||
}
|
||||
|
||||
const sessionIds = collectSessionIds(join(dataDir, "agent", "sessions"));
|
||||
const checkpointsPath = join(dataDir, "home", ".pi", "agent", "ayu", "checkpoints", "sessions");
|
||||
if (existsSync(checkpointsPath)) {
|
||||
for (const entry of readdirSync(checkpointsPath, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
const checkpointPath = join(checkpointsPath, entry.name);
|
||||
const stats = directoryStats(checkpointPath);
|
||||
const orphan = !sessionIds.has(entry.name);
|
||||
const ageMs = Math.max(0, nowMs - stats.newestMtimeMs);
|
||||
|
||||
if (orphan && (mode === "clean" || ageMs >= policy.orphanGraceMs)) {
|
||||
removeDirectory(checkpointPath, `orphan Rewind checkpoint ${entry.name}`, result);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (mode !== "status") {
|
||||
const looseObjects = countLooseGitObjects(checkpointPath);
|
||||
if (looseObjects >= policy.rewindGcLooseObjects) {
|
||||
const before = directoryStats(checkpointPath).bytes;
|
||||
try {
|
||||
compactCheckpoint(checkpointPath);
|
||||
const after = directoryStats(checkpointPath).bytes;
|
||||
result.reclaimedBytes += Math.max(0, before - after);
|
||||
result.actions.push({
|
||||
type: "compacted",
|
||||
label: `Rewind checkpoint ${entry.name}`,
|
||||
path: checkpointPath,
|
||||
bytes: Math.max(0, before - after),
|
||||
});
|
||||
} catch (error) {
|
||||
result.warnings.push(`Unable to compact ${checkpointPath}: ${error.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
export function formatBytes(bytes) {
|
||||
if (bytes >= 1024 ** 3) return `${(bytes / 1024 ** 3).toFixed(2)} GiB`;
|
||||
if (bytes >= MIB) return `${(bytes / MIB).toFixed(1)} MiB`;
|
||||
return `${(bytes / 1024).toFixed(1)} KiB`;
|
||||
}
|
||||
|
||||
export function printMaintenanceResult(result) {
|
||||
if (result.disabled) {
|
||||
console.log("[storage] automatic maintenance is disabled");
|
||||
return;
|
||||
}
|
||||
if (result.mode === "status") {
|
||||
for (const [label, stats] of Object.entries(result.status)) {
|
||||
console.log(`[storage] ${label}: ${formatBytes(stats.bytes)} (${stats.files} files)`);
|
||||
}
|
||||
}
|
||||
for (const action of result.actions) {
|
||||
const verb = action.type === "compacted" ? "compacted" : "removed";
|
||||
console.log(`[storage] ${verb} ${action.label}; reclaimed ${formatBytes(action.bytes)}`);
|
||||
}
|
||||
for (const warning of result.warnings) console.warn(`[storage] ${warning}`);
|
||||
if (result.actions.length > 0) console.log(`[storage] total reclaimed: ${formatBytes(result.reclaimedBytes)}`);
|
||||
else if (result.mode !== "auto") console.log("[storage] nothing to clean");
|
||||
}
|
||||
|
||||
export function isLocalPortListening(port, host = "127.0.0.1", timeoutMs = 250) {
|
||||
return new Promise((resolvePromise) => {
|
||||
const socket = createConnection({ port, host });
|
||||
const finish = (listening) => {
|
||||
socket.destroy();
|
||||
resolvePromise(listening);
|
||||
};
|
||||
socket.setTimeout(timeoutMs);
|
||||
socket.once("connect", () => finish(true));
|
||||
socket.once("timeout", () => finish(false));
|
||||
socket.once("error", () => finish(false));
|
||||
});
|
||||
}
|
||||
|
||||
const entryPath = process.argv[1] ? resolve(process.argv[1]) : "";
|
||||
if (entryPath === resolve(fileURLToPath(import.meta.url))) {
|
||||
const mode = process.argv[2] ?? "status";
|
||||
if (!new Set(["status", "auto", "clean"]).has(mode)) {
|
||||
console.error("Usage: node scripts/storage-maintenance.mjs [status|auto|clean]");
|
||||
process.exit(2);
|
||||
}
|
||||
if (mode !== "status" && await isLocalPortListening(30141)) {
|
||||
console.error("[storage] Pi Web is running on port 30141. Stop it first or use npm run restart.");
|
||||
process.exit(1);
|
||||
}
|
||||
printMaintenanceResult(maintainStorage({ mode }));
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { existsSync, mkdirSync, mkdtempSync, rmSync, utimesSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import test from "node:test";
|
||||
import { maintainStorage } from "./storage-maintenance.mjs";
|
||||
|
||||
function createFile(path, bytes = 2048) {
|
||||
mkdirSync(join(path, ".."), { recursive: true });
|
||||
writeFileSync(path, Buffer.alloc(bytes));
|
||||
}
|
||||
|
||||
function createCheckpoint(root, id, modifiedAt) {
|
||||
const checkpoint = join(root, "home", ".pi", "agent", "ayu", "checkpoints", "sessions", id);
|
||||
const objectDir = join(checkpoint, ".git", "objects", "ab");
|
||||
mkdirSync(objectDir, { recursive: true });
|
||||
createFile(join(objectDir, "object"));
|
||||
createFile(join(checkpoint, "manifest.json"), 64);
|
||||
for (const path of [join(objectDir, "object"), objectDir, join(checkpoint, ".git", "objects"), join(checkpoint, ".git"), join(checkpoint, "manifest.json"), checkpoint]) {
|
||||
utimesSync(path, modifiedAt, modifiedAt);
|
||||
}
|
||||
return checkpoint;
|
||||
}
|
||||
|
||||
test("automatic maintenance bounds caches and removes only expired orphan checkpoints", (t) => {
|
||||
const root = mkdtempSync(join(tmpdir(), "pi-storage-maintenance-"));
|
||||
const dataDir = join(root, "data");
|
||||
const now = new Date("2026-07-31T00:00:00Z");
|
||||
const old = new Date(now.getTime() - 8 * 24 * 60 * 60 * 1000);
|
||||
const recent = new Date(now.getTime() - 24 * 60 * 60 * 1000);
|
||||
t.after(() => rmSync(root, { recursive: true, force: true }));
|
||||
|
||||
const nextDev = join(root, "pi-web", ".next", "dev");
|
||||
const nextCache = join(root, "pi-web", ".next", "cache");
|
||||
const npmCache = join(dataDir, "cache", "npm");
|
||||
createFile(join(nextDev, "large.bin"));
|
||||
createFile(join(nextCache, "large.bin"));
|
||||
createFile(join(npmCache, "large.bin"));
|
||||
|
||||
const sessions = join(dataDir, "agent", "sessions");
|
||||
createFile(join(sessions, "live-session.jsonl"), 16);
|
||||
const liveCheckpoint = createCheckpoint(dataDir, "live-session", old);
|
||||
const expiredOrphan = createCheckpoint(dataDir, "expired-orphan", old);
|
||||
const recentOrphan = createCheckpoint(dataDir, "recent-orphan", recent);
|
||||
const compacted = [];
|
||||
|
||||
const result = maintainStorage({
|
||||
mode: "auto",
|
||||
rootDir: root,
|
||||
dataDir,
|
||||
nowMs: now.getTime(),
|
||||
env: {
|
||||
PI_STORAGE_NEXT_DEV_MAX_MB: "0.001",
|
||||
PI_STORAGE_NEXT_CACHE_MAX_MB: "0.001",
|
||||
PI_STORAGE_NPM_CACHE_MAX_MB: "0.001",
|
||||
PI_STORAGE_ORPHAN_GRACE_DAYS: "7",
|
||||
PI_STORAGE_REWIND_GC_OBJECTS: "1",
|
||||
},
|
||||
compactCheckpoint(path) {
|
||||
compacted.push(path);
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(existsSync(nextDev), false);
|
||||
assert.equal(existsSync(nextCache), false);
|
||||
assert.equal(existsSync(npmCache), false);
|
||||
assert.equal(existsSync(expiredOrphan), false);
|
||||
assert.equal(existsSync(recentOrphan), true);
|
||||
assert.equal(existsSync(liveCheckpoint), true);
|
||||
assert.deepEqual(new Set(compacted), new Set([recentOrphan, liveCheckpoint]));
|
||||
assert.ok(result.reclaimedBytes > 0);
|
||||
});
|
||||
|
||||
test("clean mode removes every orphan while preserving live checkpoints", (t) => {
|
||||
const root = mkdtempSync(join(tmpdir(), "pi-storage-clean-"));
|
||||
const dataDir = join(root, "data");
|
||||
t.after(() => rmSync(root, { recursive: true, force: true }));
|
||||
|
||||
createFile(join(dataDir, "agent", "sessions", "live-session.jsonl"), 16);
|
||||
const liveCheckpoint = createCheckpoint(dataDir, "live-session", new Date());
|
||||
const orphanCheckpoint = createCheckpoint(dataDir, "orphan-session", new Date());
|
||||
|
||||
maintainStorage({
|
||||
mode: "clean",
|
||||
rootDir: root,
|
||||
dataDir,
|
||||
env: { PI_STORAGE_REWIND_GC_OBJECTS: "999999" },
|
||||
});
|
||||
|
||||
assert.equal(existsSync(liveCheckpoint), true);
|
||||
assert.equal(existsSync(orphanCheckpoint), false);
|
||||
});
|
||||
Reference in New Issue
Block a user