From f41c143345f26ec31e2f8a268687828db2762909 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 15 Jul 2026 08:40:46 -0700 Subject: [PATCH] refactor(state): move skill upload staging to SQLite (#108346) * refactor(state): move skill upload staging to sqlite * test(skills): make archive mode assertion portable * fix(skills): fence upload finalization by lease owner * fix(skills): handle nullable lease expiry * fix(skills): bound upload install leases --- docs/refactor/database-first.md | 11 +- .../check-database-first-legacy-stores.mjs | 2 + src/commands/doctor-usage-cost-cache.test.ts | 37 + src/commands/doctor-usage-cost-cache.ts | 36 + .../server-methods/skills-upload.test.ts | 45 +- src/skills/lifecycle/upload-store.sqlite.ts | 190 ++++ .../lifecycle/upload-store.test-support.ts | 6 +- src/skills/lifecycle/upload-store.test.ts | 818 +++++++++++++---- src/skills/lifecycle/upload-store.ts | 868 ++++++++++-------- src/state/openclaw-state-db.generated.d.ts | 8 + src/state/openclaw-state-schema.generated.ts | 10 + src/state/openclaw-state-schema.sql | 10 + ...check-database-first-legacy-stores.test.ts | 13 + 13 files changed, 1458 insertions(+), 596 deletions(-) create mode 100644 src/skills/lifecycle/upload-store.sqlite.ts diff --git a/docs/refactor/database-first.md b/docs/refactor/database-first.md index 8393fb96b65e..d97ac77abcaf 100644 --- a/docs/refactor/database-first.md +++ b/docs/refactor/database-first.md @@ -1184,10 +1184,12 @@ sessionId})`; create, branch, continue, list, and fork flows live in their files. Install, doctor, update, and release smoke paths use generated completion output or profile sourcing instead of durable completion cache files. -- Gateway skill-upload staging now uses shared `skill_uploads` rows. Upload - metadata, idempotency keys, and archive bytes live in SQLite; the installer - only receives a temporary materialized archive path while an install is - running. +- Gateway skill-upload staging now uses shared `skill_uploads` and + `skill_upload_chunks` rows. Chunks stay individually transactional during + upload, then commit assembles one verified archive BLOB and removes the chunk + rows. The installer only receives a temporary materialized archive path while + an install is running. Doctor discards the retired one-hour filesystem + staging tree instead of importing transient uploads. - Subagent inline attachments no longer materialize under workspace `.openclaw/attachments/*`. The spawn path prepares SQLite VFS seed entries, inline runs seed those entries into the per-agent runtime scratch namespace, @@ -1462,6 +1464,7 @@ plugin_state_entries(plugin_id, namespace, entry_key, value_json, created_at, ex plugin_blob_entries(plugin_id, namespace, entry_key, metadata_json, blob, created_at, expires_at) media_blobs(subdir, id, content_type, size_bytes, blob, created_at, updated_at) skill_uploads(upload_id, kind, slug, force, size_bytes, sha256, actual_sha256, received_bytes, archive_blob, created_at, expires_at, committed, committed_at, idempotency_key_hash) +skill_upload_chunks(upload_id, byte_offset, size_bytes, chunk_blob) web_push_subscriptions(endpoint_hash, subscription_id, endpoint, p256dh, auth, created_at_ms, updated_at_ms) web_push_vapid_keys(key_id, public_key, private_key, subject, updated_at_ms) apns_registrations(node_id, transport, token, relay_handle, send_grant, installation_id, topic, environment, distribution, token_debug_suffix, updated_at_ms) diff --git a/scripts/check-database-first-legacy-stores.mjs b/scripts/check-database-first-legacy-stores.mjs index 582778af15fc..96b4ae2d561d 100644 --- a/scripts/check-database-first-legacy-stores.mjs +++ b/scripts/check-database-first-legacy-stores.mjs @@ -93,6 +93,7 @@ const legacyStorePatterns = [ /\btui\/last-session\.json\b/u, /\bcommitments\/commitments\.json\b/u, /\bmedia\/outgoing\/records\/[^"'`]*\.json\b/u, + /\btmp\/skill-uploads\b/u, /\b(?:crestodian|openclaw)\/rescue-pending\/[^"'`]*\.json\b/u, /\bcron\/(?:runs\/[^"'`]+\.jsonl|jobs\.json|jobs-state\.json)\b/u, /\b(?:process-leases|session-toggles|known-users|msteams-conversations|msteams-polls|msteams-sso-tokens|bot-storage|sync-store|thread-bindings|inbound-dedupe|startup-verification|storage-meta|crypto-idb-snapshot|command-deploy-cache|plugin-binding-approvals|plugins\/installs|config-health|port-guard|restart-sentinel|gateway-restart-intent|gateway-supervisor-restart-handoff)\.json\b/u, @@ -105,6 +106,7 @@ const legacyStorePatterns = [ const allowedRuntimeMigrationPaths = [ "src/commands/doctor/", + "src/commands/doctor-usage-cost-cache.ts", "src/infra/session-state-migration.ts", "src/infra/state-migrations.ts", "src/infra/state-migrations.tui-last-session.ts", diff --git a/src/commands/doctor-usage-cost-cache.test.ts b/src/commands/doctor-usage-cost-cache.test.ts index d4baa96d03f4..56af03533601 100644 --- a/src/commands/doctor-usage-cost-cache.test.ts +++ b/src/commands/doctor-usage-cost-cache.test.ts @@ -55,4 +55,41 @@ describe("legacy usage-cost cache cleanup", () => { await expect(fs.readFile(filePath, "utf-8")).resolves.toBe("x"); } }); + + it("reports legacy skill-upload staging without deleting it unless repair is enabled", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-skill-upload-doctor-")); + const uploadRoot = path.join(root, "tmp", "skill-uploads"); + const metadataPath = path.join(uploadRoot, randomUploadId(), "metadata.json"); + await fs.mkdir(path.dirname(metadataPath), { recursive: true }); + await fs.writeFile(metadataPath, "{}\n", "utf8"); + const env = { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv; + + await maybeRepairLegacyRuntimeFiles(false, env); + await expect(fs.readFile(metadataPath, "utf8")).resolves.toBe("{}\n"); + + await maybeRepairLegacyRuntimeFiles(true, env); + await expect(fs.stat(uploadRoot)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + const symlinkTest = process.platform === "win32" ? it.skip : it; + symlinkTest("removes a legacy staging symlink without touching its target", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-skill-upload-link-")); + const uploadRoot = path.join(root, "tmp", "skill-uploads"); + const external = path.join(root, "external"); + await fs.mkdir(path.dirname(uploadRoot), { recursive: true }); + await fs.mkdir(external); + await fs.writeFile(path.join(external, "keep.txt"), "keep", "utf8"); + await fs.symlink(external, uploadRoot, "dir"); + + await maybeRepairLegacyRuntimeFiles(true, { + OPENCLAW_STATE_DIR: root, + } as NodeJS.ProcessEnv); + + await expect(fs.lstat(uploadRoot)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.readFile(path.join(external, "keep.txt"), "utf8")).resolves.toBe("keep"); + }); }); + +function randomUploadId(): string { + return "11111111-1111-4111-8111-111111111111"; +} diff --git a/src/commands/doctor-usage-cost-cache.ts b/src/commands/doctor-usage-cost-cache.ts index 9b6ee0248667..382ec9ec25d8 100644 --- a/src/commands/doctor-usage-cost-cache.ts +++ b/src/commands/doctor-usage-cost-cache.ts @@ -89,10 +89,46 @@ async function maybeRemoveLegacyUsageCostCacheFiles(params: { ); } +async function maybeRemoveLegacySkillUploadTree(params: { + shouldRepair: boolean; + env?: NodeJS.ProcessEnv; + homedir?: () => string; +}): Promise { + const stateDir = resolveStateDir(params.env ?? process.env, params.homedir ?? os.homedir); + const uploadRoot = path.join(stateDir, "tmp", "skill-uploads"); + const stats = await fs.lstat(uploadRoot).catch(() => null); + if (!stats) { + return; + } + if (!params.shouldRepair) { + note( + "Legacy skill-upload staging remains. Run `openclaw doctor --fix` to discard it; active uploads now live in SQLite and must be retried.", + "Skill uploads", + ); + return; + } + try { + // Removing a symlink removes only the fixed legacy entry, never its target. + if (stats.isSymbolicLink()) { + await fs.unlink(uploadRoot); + } else { + await fs.rm(uploadRoot, { recursive: true, force: true }); + } + } catch (error) { + note(`Failed removing legacy skill-upload staging: ${String(error)}`, "Skill uploads"); + return; + } + note( + "Removed legacy skill-upload staging; unfinished transient uploads must be retried.", + "Skill uploads", + ); +} + export async function maybeRepairLegacyRuntimeFiles( shouldRepair: boolean, env?: NodeJS.ProcessEnv, ): Promise { await maybeScrubConfigAuditLog({ shouldRepair, env }); await maybeRemoveLegacyUsageCostCacheFiles({ shouldRepair, env }); + await maybeRemoveLegacySkillUploadTree({ shouldRepair, env }); } diff --git a/src/gateway/server-methods/skills-upload.test.ts b/src/gateway/server-methods/skills-upload.test.ts index 961c76c5a4be..0b03dd586fe8 100644 --- a/src/gateway/server-methods/skills-upload.test.ts +++ b/src/gateway/server-methods/skills-upload.test.ts @@ -6,6 +6,10 @@ import fs from "node:fs/promises"; import path from "node:path"; import JSZip from "jszip"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../../state/openclaw-state-db.js"; import { createOpenClawTestState, type OpenClawTestState, @@ -143,6 +147,15 @@ async function expectPathMissing(targetPath: string): Promise { } } +function skillUploadExists(stateDir: string, uploadId: string): boolean { + const { db } = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + return Boolean( + db.prepare("SELECT 1 AS found FROM skill_uploads WHERE upload_id = ?").get(uploadId), + ); +} + function expectError(result: CallResult, code: string, message: string): void { expect(result.error?.code).toBe(code); expect(result.error?.message).toBe(message); @@ -233,6 +246,7 @@ describe("skill upload gateway handlers", () => { afterEach(async () => { vi.restoreAllMocks(); + closeOpenClawStateDatabaseForTest(); await Promise.all([ ...tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })), ...testStates.splice(0).map((state) => state.cleanup()), @@ -300,7 +314,8 @@ describe("skill upload gateway handlers", () => { fs.readFile(path.join(workspaceDir, "skills", "uploaded-demo", "SKILL.md"), "utf8"), ).resolves.toContain("Uploaded Demo"); await expectPathMissing(path.join(workspaceDir, "skills", "archive-internal-name")); - await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads", uploadId)); + expect(skillUploadExists(stateDir, uploadId)).toBe(false); + await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads")); const status = await call(handlers, "skills.status", {}); expect(status.ok).toBe(true); @@ -380,7 +395,7 @@ describe("skill upload gateway handlers", () => { expect(install.ok).toBe(false); expectError(install, "INVALID_REQUEST", "install sha256 does not match uploaded archive"); - await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads", upload.uploadId)); + expect(skillUploadExists(stateDir, upload.uploadId)).toBe(false); }); it("rejects expired committed uploads through skills.install", async () => { @@ -389,16 +404,9 @@ describe("skill upload gateway handlers", () => { archive: await makeSkillArchive({}), slug: "expired-skill", }); - const metadataPath = path.join( - stateDir, - "tmp", - "skill-uploads", - upload.uploadId, - "metadata.json", - ); - const metadata = JSON.parse(await fs.readFile(metadataPath, "utf8")) as { expiresAt: number }; - metadata.expiresAt = Date.now() - 1; - await fs.writeFile(metadataPath, `${JSON.stringify(metadata, null, 2)}\n`, "utf8"); + openOpenClawStateDatabase({ env: { ...process.env, OPENCLAW_STATE_DIR: stateDir } }) + .db.prepare("UPDATE skill_uploads SET expires_at = ? WHERE upload_id = ?") + .run(Date.now() - 1, upload.uploadId); const install = await call(handlers, "skills.install", { source: "upload", @@ -408,7 +416,7 @@ describe("skill upload gateway handlers", () => { expect(install.ok).toBe(false); expectError(install, "INVALID_REQUEST", "upload has expired"); - await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads", upload.uploadId)); + expect(skillUploadExists(stateDir, upload.uploadId)).toBe(false); }); it("rejects invalid slugs, missing SKILL.md, and archive traversal", async () => { @@ -433,7 +441,7 @@ describe("skill upload gateway handlers", () => { expect(missingInstall.ok).toBe(false); expect(missingInstall.error?.code).toBe("INVALID_REQUEST"); expect(missingInstall.error?.message).toContain("SKILL.md"); - await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads", missingSkill.uploadId)); + expect(skillUploadExists(stateDir, missingSkill.uploadId)).toBe(false); const legacyMarker = await uploadArchive(handlers, { archive: await makeSkillArchive({ @@ -450,7 +458,7 @@ describe("skill upload gateway handlers", () => { expect(legacyMarkerInstall.ok).toBe(false); expect(legacyMarkerInstall.error?.code).toBe("INVALID_REQUEST"); expect(legacyMarkerInstall.error?.message).toContain("SKILL.md"); - await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads", legacyMarker.uploadId)); + expect(skillUploadExists(stateDir, legacyMarker.uploadId)).toBe(false); const traversal = await uploadArchive(handlers, { archive: await makeSkillArchive({ traversal: true }), @@ -498,7 +506,7 @@ describe("skill upload gateway handlers", () => { expect(scanInput.origin?.type).toBe("upload"); expect(scanInput.origin?.uploadId).toBe(upload.uploadId); expect(scanInput.skillName).toBe("scan-blocked"); - await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads", upload.uploadId)); + expect(skillUploadExists(stateDir, upload.uploadId)).toBe(false); }); it("preserves existing installs unless force was bound at begin", async () => { @@ -535,7 +543,7 @@ describe("skill upload gateway handlers", () => { expect(blockedInstall.ok).toBe(false); expect(blockedInstall.error?.code).toBe("INVALID_REQUEST"); expect(blockedInstall.error?.message).toContain("already exists"); - await expectPathMissing(path.join(stateDir, "tmp", "skill-uploads", blocked.uploadId)); + expect(skillUploadExists(stateDir, blocked.uploadId)).toBe(false); const forced = await uploadArchive(handlers, { archive: await makeSkillArchive({ @@ -602,7 +610,6 @@ describe("skill upload gateway handlers", () => { await expect( fs.readFile(path.join(workspaceDir, "skills", "rollback-demo", "SKILL.md"), "utf8"), ).resolves.toContain("first version"); - const uploadStat = await fs.stat(path.join(stateDir, "tmp", "skill-uploads", forced.uploadId)); - expect(uploadStat.isDirectory()).toBe(true); + expect(skillUploadExists(stateDir, forced.uploadId)).toBe(true); }); }); diff --git a/src/skills/lifecycle/upload-store.sqlite.ts b/src/skills/lifecycle/upload-store.sqlite.ts new file mode 100644 index 000000000000..8226b55fdbe5 --- /dev/null +++ b/src/skills/lifecycle/upload-store.sqlite.ts @@ -0,0 +1,190 @@ +// SQLite ownership helpers for Gateway skill-upload staging. +import type { DatabaseSync } from "node:sqlite"; +import type { Kysely, Selectable } from "kysely"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "../../infra/kysely-sync.js"; +import type { + DB as OpenClawStateDatabase, + SkillUploads, +} from "../../state/openclaw-state-db.generated.js"; +import { + openOpenClawStateDatabase, + runOpenClawStateWriteTransaction, + type OpenClawStateDatabaseOptions, +} from "../../state/openclaw-state-db.js"; + +export const SKILL_UPLOAD_LEASE_SCOPE = "skill-upload-install"; + +export type SkillUploadDatabase = Pick< + OpenClawStateDatabase, + "skill_upload_chunks" | "skill_uploads" | "state_leases" +>; +export type SkillUploadRow = Selectable; + +export function resolveSkillUploadDatabaseOptions(options: { + env?: NodeJS.ProcessEnv; + path?: string; +}): OpenClawStateDatabaseOptions { + return { + ...(options.env ? { env: options.env } : {}), + ...(options.path ? { path: options.path } : {}), + }; +} + +export function openSkillUploadDatabase(options: OpenClawStateDatabaseOptions) { + const database = openOpenClawStateDatabase(options); + return { + database, + kysely: getNodeSqliteKysely(database.db), + }; +} + +export function readSkillUploadRow( + uploadId: string, + options: OpenClawStateDatabaseOptions, +): SkillUploadRow | undefined { + const { database, kysely } = openSkillUploadDatabase(options); + return executeSqliteQueryTakeFirstSync( + database.db, + kysely.selectFrom("skill_uploads").selectAll().where("upload_id", "=", uploadId), + ); +} + +export function deleteSkillUploadState( + db: DatabaseSync, + kysely: Kysely, + uploadId: string, +): void { + executeSqliteQuerySync( + db, + kysely + .deleteFrom("state_leases") + .where("scope", "=", SKILL_UPLOAD_LEASE_SCOPE) + .where("lease_key", "=", uploadId), + ); + executeSqliteQuerySync(db, kysely.deleteFrom("skill_uploads").where("upload_id", "=", uploadId)); +} + +export function deleteOwnedSkillUpload( + uploadId: string, + owner: string, + nowMs: number, + options: OpenClawStateDatabaseOptions, +): "deleted" | "missing" | "not-owner" { + return runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + const upload = executeSqliteQueryTakeFirstSync( + db, + kysely.selectFrom("skill_uploads").select("upload_id").where("upload_id", "=", uploadId), + ); + if (!upload) { + return "missing"; + } + const lease = executeSqliteQueryTakeFirstSync( + db, + kysely + .selectFrom("state_leases") + .select(["owner", "expires_at"]) + .where("scope", "=", SKILL_UPLOAD_LEASE_SCOPE) + .where("lease_key", "=", uploadId), + ); + if (!lease || lease.owner !== owner || lease.expires_at === null || lease.expires_at <= nowMs) { + return "not-owner"; + } + deleteSkillUploadState(db, kysely, uploadId); + return "deleted"; + }, options); +} + +export function hasLiveSkillUploadInstallLease( + db: DatabaseSync, + kysely: Kysely, + uploadId: string, + nowMs: number, +): boolean { + return Boolean( + executeSqliteQueryTakeFirstSync( + db, + kysely + .selectFrom("state_leases") + .select("lease_key") + .where("scope", "=", SKILL_UPLOAD_LEASE_SCOPE) + .where("lease_key", "=", uploadId) + .where("expires_at", ">", nowMs), + ), + ); +} + +export function deleteExpiredSkillUploadUnlessLeased(params: { + uploadId: string; + nowMs: number; + options: OpenClawStateDatabaseOptions; +}): "active" | "deleted" | "leased" | "missing" { + return runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + const row = executeSqliteQueryTakeFirstSync( + db, + kysely + .selectFrom("skill_uploads") + .select("expires_at") + .where("upload_id", "=", params.uploadId), + ); + if (!row) { + return "missing"; + } + if (row.expires_at > params.nowMs) { + return "active"; + } + if (hasLiveSkillUploadInstallLease(db, kysely, params.uploadId, params.nowMs)) { + return "leased"; + } + deleteSkillUploadState(db, kysely, params.uploadId); + return "deleted"; + }, params.options); +} + +export function renewSkillUploadInstallLease(params: { + uploadId: string; + owner: string; + heartbeatAt: number; + expiresAt: number; + options: OpenClawStateDatabaseOptions; +}): boolean { + return runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + return ( + executeSqliteQuerySync( + db, + kysely + .updateTable("state_leases") + .set({ + heartbeat_at: params.heartbeatAt, + expires_at: params.expiresAt, + updated_at: params.heartbeatAt, + }) + .where("scope", "=", SKILL_UPLOAD_LEASE_SCOPE) + .where("lease_key", "=", params.uploadId) + .where("owner", "=", params.owner) + .where("expires_at", ">", params.heartbeatAt), + ).numAffectedRows === 1n + ); + }, params.options); +} + +export function readSkillUploadArchiveChunks( + uploadId: string, + options: OpenClawStateDatabaseOptions, +): Array<{ byte_offset: number; size_bytes: number; chunk_blob: Uint8Array }> { + const { database, kysely } = openSkillUploadDatabase(options); + return executeSqliteQuerySync( + database.db, + kysely + .selectFrom("skill_upload_chunks") + .select(["byte_offset", "size_bytes", "chunk_blob"]) + .where("upload_id", "=", uploadId) + .orderBy("byte_offset", "asc"), + ).rows; +} diff --git a/src/skills/lifecycle/upload-store.test-support.ts b/src/skills/lifecycle/upload-store.test-support.ts index 1c1cfb853d98..1e112b9436b3 100644 --- a/src/skills/lifecycle/upload-store.test-support.ts +++ b/src/skills/lifecycle/upload-store.test-support.ts @@ -3,8 +3,12 @@ import "./upload-store.js"; type SkillUploadStoreTestApi = { createSkillUploadStore(options?: { - rootDir?: string; + env?: NodeJS.ProcessEnv; + installLeaseHeartbeatMs?: number; + installLeaseMs?: number; now?: () => number; + path?: string; + tempRootDir?: string; ttlMs?: number; }): SkillUploadStore; }; diff --git a/src/skills/lifecycle/upload-store.test.ts b/src/skills/lifecycle/upload-store.test.ts index e75f7124aa0d..18315ee61963 100644 --- a/src/skills/lifecycle/upload-store.test.ts +++ b/src/skills/lifecycle/upload-store.test.ts @@ -1,13 +1,41 @@ -// Upload store tests cover staged skill archive persistence and cleanup. +// Upload store tests cover SQLite staging, integrity, concurrency, and cleanup. import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { MAX_DATE_TIMESTAMP_MS } from "@openclaw/normalization-core/number-coercion"; -import { afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../../state/openclaw-state-db.js"; import { SkillUploadRequestError } from "./upload-store.js"; +import { + deleteExpiredSkillUploadUnlessLeased, + renewSkillUploadInstallLease, +} from "./upload-store.sqlite.js"; import { createSkillUploadStore } from "./upload-store.test-support.js"; +type ReadSkillUploadArchiveChunks = + typeof import("./upload-store.sqlite.js").readSkillUploadArchiveChunks; + +const uploadSqliteMocks = vi.hoisted(() => ({ + defaultReadSkillUploadArchiveChunks: undefined as ReadSkillUploadArchiveChunks | undefined, + readSkillUploadArchiveChunks: vi.fn(), +})); + +vi.mock("./upload-store.sqlite.js", async (importOriginal) => { + const actual = await importOriginal(); + uploadSqliteMocks.defaultReadSkillUploadArchiveChunks = actual.readSkillUploadArchiveChunks; + uploadSqliteMocks.readSkillUploadArchiveChunks.mockImplementation( + actual.readSkillUploadArchiveChunks, + ); + return { + ...actual, + readSkillUploadArchiveChunks: uploadSqliteMocks.readSkillUploadArchiveChunks, + }; +}); + const ACTIVE_UPLOAD_LIMIT = 32; let tempDirs: string[] = []; @@ -18,6 +46,66 @@ async function makeTempDir(): Promise { return dir; } +async function makeStore(options?: { + installLeaseHeartbeatMs?: number; + installLeaseMs?: number; + now?: () => number; + ttlMs?: number; +}) { + const root = await makeTempDir(); + const databasePath = path.join(root, "openclaw.sqlite"); + return { + root, + databasePath, + store: createSkillUploadStore({ + path: databasePath, + tempRootDir: root, + ...options, + }), + }; +} + +function stateDatabase(databasePath: string) { + return openOpenClawStateDatabase({ path: databasePath }).db; +} + +function uploadCount(databasePath: string): number { + return ( + stateDatabase(databasePath).prepare("SELECT count(*) AS count FROM skill_uploads").get() as { + count: number; + } + ).count; +} + +function uploadExists(databasePath: string, uploadId: string): boolean { + return Boolean( + stateDatabase(databasePath) + .prepare("SELECT 1 AS found FROM skill_uploads WHERE upload_id = ?") + .get(uploadId), + ); +} + +function chunkCount(databasePath: string, uploadId?: string): number { + const row = uploadId + ? stateDatabase(databasePath) + .prepare("SELECT count(*) AS count FROM skill_upload_chunks WHERE upload_id = ?") + .get(uploadId) + : stateDatabase(databasePath) + .prepare("SELECT count(*) AS count FROM skill_upload_chunks") + .get(); + return (row as { count: number }).count; +} + +function installLeaseCount(databasePath: string, uploadId: string): number { + return ( + stateDatabase(databasePath) + .prepare( + "SELECT count(*) AS count FROM state_leases WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .get(uploadId) as { count: number } + ).count; +} + function sha256(bytes: Buffer): string { return createHash("sha256").update(bytes).digest("hex"); } @@ -50,40 +138,33 @@ async function expectUploadError( } async function expectMissingPath(targetPath: string): Promise { - try { - await fs.stat(targetPath); - } catch (err) { - expect((err as { code?: unknown }).code).toBe("ENOENT"); - return; - } - throw new Error(`expected missing path: ${targetPath}`); + await expect(fs.stat(targetPath)).rejects.toMatchObject({ code: "ENOENT" }); } describe("skill upload store", () => { let activeUploadLimitError: unknown; + let activeLimitRoot: string | undefined; beforeAll(async () => { - const rootDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-skill-upload-store-")); + activeLimitRoot = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-skill-upload-limit-")); + const store = createSkillUploadStore({ + path: path.join(activeLimitRoot, "openclaw.sqlite"), + tempRootDir: activeLimitRoot, + }); + for (let i = 0; i < ACTIVE_UPLOAD_LIMIT; i += 1) { + await store.begin({ kind: "skill-archive", slug: `active-${i}`, sizeBytes: 1 }); + } try { - const store = createSkillUploadStore({ rootDir }); - for (let i = 0; i < ACTIVE_UPLOAD_LIMIT; i += 1) { - await store.begin({ - kind: "skill-archive", - slug: `active-${i}`, - sizeBytes: 1, - }); - } - try { - await store.begin({ - kind: "skill-archive", - slug: "too-many", - sizeBytes: 1, - }); - } catch (err) { - activeUploadLimitError = err; - } - } finally { - await fs.rm(rootDir, { recursive: true, force: true }); + await store.begin({ kind: "skill-archive", slug: "too-many", sizeBytes: 1 }); + } catch (err) { + activeUploadLimitError = err; + } + }); + + afterAll(async () => { + closeOpenClawStateDatabaseForTest(); + if (activeLimitRoot) { + await fs.rm(activeLimitRoot, { recursive: true, force: true }); } }); @@ -92,14 +173,18 @@ describe("skill upload store", () => { }); afterEach(async () => { + uploadSqliteMocks.readSkillUploadArchiveChunks.mockReset(); + uploadSqliteMocks.readSkillUploadArchiveChunks.mockImplementation( + uploadSqliteMocks.defaultReadSkillUploadArchiveChunks!, + ); + closeOpenClawStateDatabaseForTest(); await Promise.all( tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })), ); }); - it("stores chunks and commits an archive with sha verification", async () => { - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ rootDir }); + it("stores chunks, commits one archive blob, and materializes only for the action", async () => { + const { root, databasePath, store } = await makeStore(); const archive = Buffer.from("zip-bytes"); const digest = sha256(archive); const begin = await store.begin({ @@ -116,7 +201,6 @@ describe("skill upload store", () => { sha256: digest, idempotencyKey: "same-upload", }); - expect(repeated.uploadId).toBe(begin.uploadId); await store.chunk({ @@ -124,6 +208,7 @@ describe("skill upload store", () => { offset: 0, dataBase64: archive.subarray(0, 3).toString("base64"), }); + expect(chunkCount(databasePath, begin.uploadId)).toBe(1); const chunk = await store.chunk({ uploadId: begin.uploadId, offset: 3, @@ -132,39 +217,49 @@ describe("skill upload store", () => { expect(chunk.receivedBytes).toBe(archive.length); const commit = await store.commit({ uploadId: begin.uploadId, sha256: digest }); - expect(commit.uploadId).toBe(begin.uploadId); - expect(commit.receivedBytes).toBe(archive.length); - expect(commit.sha256).toBe(digest); + expect(commit).toMatchObject({ + uploadId: begin.uploadId, + receivedBytes: archive.length, + sha256: digest, + }); + expect(chunkCount(databasePath, begin.uploadId)).toBe(0); + const persisted = stateDatabase(databasePath) + .prepare( + "SELECT archive_blob, committed, actual_sha256 FROM skill_uploads WHERE upload_id = ?", + ) + .get(begin.uploadId) as { + archive_blob: Uint8Array; + committed: number; + actual_sha256: string; + }; + expect(Buffer.from(persisted.archive_blob)).toEqual(archive); + expect(persisted).toMatchObject({ committed: 1, actual_sha256: digest }); + let materializedPath = ""; const record = await store.withCommittedUpload(begin.uploadId, async (committedRecord) => { + materializedPath = committedRecord.archivePath; + expect(await fs.readFile(materializedPath)).toEqual(archive); + if (process.platform !== "win32") { + expect((await fs.stat(materializedPath)).mode & 0o777).toBe(0o600); + } return committedRecord; }); - expect(record.uploadId).toBe(begin.uploadId); - expect(record.slug).toBe("demo-skill"); - expect(record.force).toBe(false); - expect(record.receivedBytes).toBe(archive.length); - expect(record.actualSha256).toBe(digest); - expect(record.committed).toBe(true); - await expectUploadError( - store.chunk({ - uploadId: begin.uploadId, - offset: archive.length, - dataBase64: Buffer.from("x").toString("base64"), - }), - "upload is already committed", - ); + expect(record).toMatchObject({ + uploadId: begin.uploadId, + slug: "demo-skill", + force: false, + receivedBytes: archive.length, + actualSha256: digest, + committed: true, + }); + await expectMissingPath(materializedPath); + await expectMissingPath(path.join(root, "tmp", "skill-uploads")); }); it("rejects traversal slugs and missing uploads", async () => { - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ rootDir }); - + const { store } = await makeStore(); await expectUploadError( - store.begin({ - kind: "skill-archive", - slug: "../escape", - sizeBytes: 1, - }), + store.begin({ kind: "skill-archive", slug: "../escape", sizeBytes: 1 }), "Invalid skill slug: ../escape", ); await expectUploadError( @@ -174,15 +269,13 @@ describe("skill upload store", () => { }); it("rejects offset, size, and sha mismatches", async () => { - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ rootDir }); + const { store } = await makeStore(); const archive = Buffer.from("abc"); const begin = await store.begin({ kind: "skill-archive", slug: "demo-skill", sizeBytes: archive.length, }); - await expectUploadError( store.chunk({ uploadId: begin.uploadId, @@ -225,37 +318,160 @@ describe("skill upload store", () => { ); }); - it("truncates stale archive tails before retrying a chunk at the recorded offset", async () => { - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ rootDir }); + it("resumes a multi-chunk upload from a new store instance", async () => { + const { databasePath, root, store } = await makeStore(); const archive = Buffer.from("abcdef"); const begin = await store.begin({ kind: "skill-archive", - slug: "retry-skill", + slug: "resume-skill", sizeBytes: archive.length, }); - await store.chunk({ uploadId: begin.uploadId, offset: 0, dataBase64: archive.subarray(0, 3).toString("base64"), }); - const archivePath = path.join(rootDir, begin.uploadId, "archive.zip"); - await fs.appendFile(archivePath, Buffer.from("stale-tail")); - await store.chunk({ + + const reopened = createSkillUploadStore({ path: databasePath, tempRootDir: root }); + await reopened.chunk({ uploadId: begin.uploadId, offset: 3, dataBase64: archive.subarray(3).toString("base64"), }); + await expect( + reopened.commit({ uploadId: begin.uploadId, sha256: sha256(archive) }), + ).resolves.toMatchObject({ sha256: sha256(archive) }); + }); - await expect(fs.readFile(archivePath)).resolves.toEqual(archive); - const commit = await store.commit({ uploadId: begin.uploadId, sha256: sha256(archive) }); - expect(commit.sha256).toBe(sha256(archive)); + it("keeps large chunks separate until one final archive write", async () => { + const { databasePath, store } = await makeStore(); + const firstChunk = Buffer.alloc(4 * 1024 * 1024, 0x61); + const secondChunk = Buffer.alloc(4 * 1024 * 1024, 0x62); + const archive = Buffer.concat([firstChunk, secondChunk]); + const begin = await store.begin({ + kind: "skill-archive", + slug: "large-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: begin.uploadId, + offset: 0, + dataBase64: firstChunk.toString("base64"), + }); + await store.chunk({ + uploadId: begin.uploadId, + offset: firstChunk.length, + dataBase64: secondChunk.toString("base64"), + }); + const staged = stateDatabase(databasePath) + .prepare("SELECT length(archive_blob) AS bytes FROM skill_uploads WHERE upload_id = ?") + .get(begin.uploadId) as { bytes: number }; + expect(staged.bytes).toBe(0); + expect(chunkCount(databasePath, begin.uploadId)).toBe(2); + + await store.commit({ uploadId: begin.uploadId, sha256: sha256(archive) }); + const committed = stateDatabase(databasePath) + .prepare("SELECT length(archive_blob) AS bytes FROM skill_uploads WHERE upload_id = ?") + .get(begin.uploadId) as { bytes: number }; + expect(committed.bytes).toBe(archive.length); + expect(chunkCount(databasePath, begin.uploadId)).toBe(0); + }); + + it("uses the expiry and idempotency indexes", async () => { + const { databasePath } = await makeStore(); + const db = stateDatabase(databasePath); + const expiryPlan = db + .prepare("EXPLAIN QUERY PLAN SELECT upload_id FROM skill_uploads WHERE expires_at <= ?") + .all(Date.now()) as Array<{ detail: string }>; + const idempotencyPlan = db + .prepare( + "EXPLAIN QUERY PLAN SELECT upload_id FROM skill_uploads WHERE idempotency_key_hash = ?", + ) + .all("hash") as Array<{ detail: string }>; + expect(expiryPlan.map((row) => row.detail).join("\n")).toContain("idx_skill_uploads_expiry"); + expect(idempotencyPlan.map((row) => row.detail).join("\n")).toMatch( + /idx_skill_uploads_idempotency|sqlite_autoindex_skill_uploads/u, + ); + }); + + it("accepts exactly one concurrent chunk at the same offset", async () => { + const { databasePath, root, store } = await makeStore(); + const begin = await store.begin({ + kind: "skill-archive", + slug: "concurrent-skill", + sizeBytes: 2, + }); + const secondStore = createSkillUploadStore({ path: databasePath, tempRootDir: root }); + const results = await Promise.allSettled([ + store.chunk({ + uploadId: begin.uploadId, + offset: 0, + dataBase64: Buffer.from("a").toString("base64"), + }), + secondStore.chunk({ + uploadId: begin.uploadId, + offset: 0, + dataBase64: Buffer.from("b").toString("base64"), + }), + ]); + expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(1); + const rejected = results.find((result) => result.status === "rejected"); + expect(rejected).toMatchObject({ + status: "rejected", + reason: expect.objectContaining({ message: "upload offset mismatch: expected 1, got 0" }), + }); + expect(chunkCount(databasePath, begin.uploadId)).toBe(1); + }); + + it("creates one row for concurrent idempotent begins and rejects conflicts", async () => { + const { databasePath, root, store } = await makeStore(); + const secondStore = createSkillUploadStore({ path: databasePath, tempRootDir: root }); + const params = { + kind: "skill-archive" as const, + slug: "idem-skill", + sizeBytes: 3, + idempotencyKey: "same-key", + }; + const [first, second] = await Promise.all([store.begin(params), secondStore.begin(params)]); + expect(second.uploadId).toBe(first.uploadId); + expect(uploadCount(databasePath)).toBe(1); + await expectUploadError( + store.begin({ ...params, slug: "different-skill" }), + "idempotencyKey conflicts with a different upload", + ); + }); + + it("keeps the optional begin sha immutable across commit retries", async () => { + const { databasePath, store } = await makeStore(); + const archive = Buffer.from("abc"); + const digest = sha256(archive); + const params = { + kind: "skill-archive" as const, + slug: "late-sha-skill", + sizeBytes: archive.length, + idempotencyKey: "late-sha-key", + }; + const begin = await store.begin(params); + await store.chunk({ + uploadId: begin.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: begin.uploadId, sha256: digest }); + + await expect(store.begin(params)).resolves.toMatchObject({ + uploadId: begin.uploadId, + receivedBytes: archive.length, + }); + expect( + stateDatabase(databasePath) + .prepare("SELECT sha256, actual_sha256 FROM skill_uploads WHERE upload_id = ?") + .get(begin.uploadId), + ).toMatchObject({ sha256: null, actual_sha256: digest }); }); it("rejects idempotent commit when committed metadata is missing the actual sha", async () => { - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ rootDir }); + const { databasePath, store } = await makeStore(); const archive = Buffer.from("abc"); const begin = await store.begin({ kind: "skill-archive", @@ -268,17 +484,56 @@ describe("skill upload store", () => { dataBase64: archive.toString("base64"), }); await store.commit({ uploadId: begin.uploadId }); - const metadataPath = path.join(rootDir, begin.uploadId, "metadata.json"); - const metadata = JSON.parse(await fs.readFile(metadataPath, "utf8")) as Record; - delete metadata.actualSha256; - await fs.writeFile(metadataPath, `${JSON.stringify(metadata, null, 2)}\n`, "utf8"); - + stateDatabase(databasePath) + .prepare("UPDATE skill_uploads SET actual_sha256 = NULL WHERE upload_id = ?") + .run(begin.uploadId); await expectUploadError( store.commit({ uploadId: begin.uploadId }), "committed upload is missing sha256", ); }); + it("returns the committed result when another process deletes chunks first", async () => { + const { databasePath, store } = await makeStore(); + const archive = Buffer.from("concurrent-commit"); + const digest = sha256(archive); + const begin = await store.begin({ + kind: "skill-archive", + slug: "concurrent-commit-skill", + sizeBytes: archive.length, + sha256: digest, + }); + await store.chunk({ + uploadId: begin.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + uploadSqliteMocks.readSkillUploadArchiveChunks.mockImplementationOnce((uploadId, options) => { + const db = stateDatabase(databasePath); + db.exec("BEGIN IMMEDIATE"); + try { + db.prepare( + "UPDATE skill_uploads SET archive_blob = ?, actual_sha256 = ?, committed = 1, committed_at = ? WHERE upload_id = ?", + ).run(archive, digest, Date.now(), uploadId); + db.prepare("DELETE FROM skill_upload_chunks WHERE upload_id = ?").run(uploadId); + db.exec("COMMIT"); + } catch (err) { + db.exec("ROLLBACK"); + throw err; + } + return uploadSqliteMocks.defaultReadSkillUploadArchiveChunks!(uploadId, options); + }); + + await expect(store.commit({ uploadId: begin.uploadId, sha256: digest })).resolves.toMatchObject( + { + uploadId: begin.uploadId, + receivedBytes: archive.length, + sha256: digest, + }, + ); + expect(chunkCount(databasePath, begin.uploadId)).toBe(0); + }); + it("limits active uploads", async () => { await expectUploadError( Promise.reject(toLintErrorObject(activeUploadLimitError, "Non-Error rejection")), @@ -287,74 +542,27 @@ describe("skill upload store", () => { }); it("rejects new uploads when the clock cannot produce a valid expiry", async () => { - const rootDir = await makeTempDir(); - const invalidClockStore = createSkillUploadStore({ - rootDir, - now: () => Number.NaN, - }); + const invalid = await makeStore({ now: () => Number.NaN }); await expectUploadError( - invalidClockStore.begin({ - kind: "skill-archive", - slug: "invalid-clock", - sizeBytes: 1, - }), + invalid.store.begin({ kind: "skill-archive", slug: "invalid-clock", sizeBytes: 1 }), "invalid upload expiry", ); - - const overflowStore = createSkillUploadStore({ - rootDir, - now: () => MAX_DATE_TIMESTAMP_MS, - }); + const overflow = await makeStore({ now: () => MAX_DATE_TIMESTAMP_MS }); await expectUploadError( - overflowStore.begin({ - kind: "skill-archive", - slug: "overflow-clock", - sizeBytes: 1, - }), + overflow.store.begin({ kind: "skill-archive", slug: "overflow-clock", sizeBytes: 1 }), "invalid upload expiry", ); }); - it("does not count uploads with invalid stored expiry as active", async () => { - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ rootDir }); - const begin = await store.begin({ - kind: "skill-archive", - slug: "invalid-expiry", - sizeBytes: 1, - idempotencyKey: "invalid-expiry", - }); - const metadataPath = path.join(rootDir, begin.uploadId, "metadata.json"); - const metadata = JSON.parse(await fs.readFile(metadataPath, "utf8")) as Record; - metadata.expiresAt = null; - await fs.writeFile(metadataPath, `${JSON.stringify(metadata, null, 2)}\n`, "utf8"); - - const repeated = await store.begin({ - kind: "skill-archive", - slug: "invalid-expiry", - sizeBytes: 1, - idempotencyKey: "invalid-expiry", - }); - - expect(repeated.uploadId).not.toBe(begin.uploadId); - await expectMissingPath(path.join(rootDir, begin.uploadId)); - }); - it("expires unfinished and committed uploads", async () => { let now = 1000; - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ - rootDir, - ttlMs: 10, - now: () => now, - }); + const { databasePath, store } = await makeStore({ ttlMs: 10, now: () => now }); const archive = Buffer.from("abc"); const begin = await store.begin({ kind: "skill-archive", slug: "demo-skill", sizeBytes: archive.length, }); - now = 1011; await expectUploadError( store.chunk({ @@ -364,6 +572,7 @@ describe("skill upload store", () => { }), "upload has expired", ); + expect(uploadCount(databasePath)).toBe(0); now = 2000; const committed = await store.begin({ @@ -382,16 +591,12 @@ describe("skill upload store", () => { store.withCommittedUpload(committed.uploadId, async (record) => record), "upload has expired", ); + expect(uploadCount(databasePath)).toBe(0); }); - it("does not sweep committed uploads while an install holds the upload lock", async () => { + it("does not sweep an upload while an install holds its lease", async () => { let now = 1000; - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ - rootDir, - ttlMs: 10, - now: () => now, - }); + const { databasePath, store } = await makeStore({ ttlMs: 10, now: () => now }); const archive = Buffer.from("abc"); const committed = await store.begin({ kind: "skill-archive", @@ -413,38 +618,75 @@ describe("skill upload store", () => { return true; }); await entered.promise; - now = 1011; - const sweep = store.begin({ + const sweep = store.begin({ kind: "skill-archive", slug: "sweep-trigger", sizeBytes: 1 }); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(uploadCount(databasePath)).toBe(1); + + now = 5000; + release.resolve(); + await expect(pinned).resolves.toBe(true); + await expect(sweep).resolves.toMatchObject({ expiresAt: 5010 }); + expect(uploadCount(databasePath)).toBe(1); + }); + + it("rechecks chunk expiry after cleanup waits on another install", async () => { + let now = 1000; + const { databasePath, store } = await makeStore({ ttlMs: 10, now: () => now }); + const archive = Buffer.from("abc"); + const pinnedUpload = await store.begin({ kind: "skill-archive", - slug: "sweep-trigger", - sizeBytes: 1, + slug: "blocking-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: pinnedUpload.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: pinnedUpload.uploadId }); + now = 1005; + const pending = await store.begin({ + kind: "skill-archive", + slug: "waiting-skill", + sizeBytes: archive.length, + }); + + const entered = deferred(); + const release = deferred(); + const pinned = store.withCommittedUpload(pinnedUpload.uploadId, async () => { + entered.resolve(); + await release.promise; + }); + await entered.promise; + now = 1011; + const chunk = store.chunk({ + uploadId: pending.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), }); await new Promise((resolve) => { setImmediate(resolve); }); - expect((await fs.stat(path.join(rootDir, committed.uploadId))).isDirectory()).toBe(true); - + now = 1020; release.resolve(); - await expect(pinned).resolves.toBe(true); - await sweep; - await expectMissingPath(path.join(rootDir, committed.uploadId)); + await pinned; + await expectUploadError(chunk, "upload has expired"); + expect(uploadExists(databasePath, pending.uploadId)).toBe(false); }); - it("does not remove expired idempotent uploads while an install holds the upload lock", async () => { - let now = 1000; - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ - rootDir, - ttlMs: 10, - now: () => now, + it("renews the install lease and preserves an expired leased upload", async () => { + const { databasePath, store } = await makeStore({ + installLeaseHeartbeatMs: 10, + installLeaseMs: 100, }); const archive = Buffer.from("abc"); const committed = await store.begin({ kind: "skill-archive", - slug: "idempotent-skill", + slug: "heartbeat-skill", sizeBytes: archive.length, - idempotencyKey: "same-upload", }); await store.chunk({ uploadId: committed.uploadId, @@ -458,66 +700,236 @@ describe("skill upload store", () => { const pinned = store.withCommittedUpload(committed.uploadId, async () => { entered.resolve(); await release.promise; - return true; }); await entered.promise; - - now = 1011; - const repeated = store.begin({ - kind: "skill-archive", - slug: "idempotent-skill", - sizeBytes: archive.length, - idempotencyKey: "same-upload", - }); + const db = stateDatabase(databasePath); + const initialHeartbeat = ( + db + .prepare( + "SELECT heartbeat_at FROM state_leases WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .get(committed.uploadId) as { heartbeat_at: number } + ).heartbeat_at; await new Promise((resolve) => { - setImmediate(resolve); + setTimeout(resolve, 40); }); - expect((await fs.stat(path.join(rootDir, committed.uploadId))).isDirectory()).toBe(true); + const renewedHeartbeat = ( + db + .prepare( + "SELECT heartbeat_at FROM state_leases WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .get(committed.uploadId) as { heartbeat_at: number } + ).heartbeat_at; + expect(renewedHeartbeat).toBeGreaterThan(initialHeartbeat); + + db.prepare("UPDATE skill_uploads SET expires_at = ? WHERE upload_id = ?").run( + Date.now() - 1, + committed.uploadId, + ); + expect( + deleteExpiredSkillUploadUnlessLeased({ + uploadId: committed.uploadId, + nowMs: Date.now(), + options: { path: databasePath }, + }), + ).toBe("leased"); + expect(uploadCount(databasePath)).toBe(1); + expect(installLeaseCount(databasePath, committed.uploadId)).toBe(1); release.resolve(); - await expect(pinned).resolves.toBe(true); - const next = await repeated; - expect(next.uploadId).not.toBe(committed.uploadId); - await expectMissingPath(path.join(rootDir, committed.uploadId)); + await pinned; + expect(installLeaseCount(databasePath, committed.uploadId)).toBe(0); }); - it("clears the orphaned idempotency pointer when re-begin hits corrupt metadata before the active cap throws", async () => { - const rootDir = await makeTempDir(); - const store = createSkillUploadStore({ rootDir }); - const idempotencyKey = "orphan-pointer-key"; - - // begin with an idempotency key → writes the upload plus its idempotency pointer - const first = await store.begin({ - kind: "skill-archive", - slug: "demo-skill", - sizeBytes: 8, - idempotencyKey, + it("starts install lease expiry from the claim time", async () => { + let now = 1000; + const { databasePath, store } = await makeStore({ + installLeaseHeartbeatMs: 1000, + installLeaseMs: 100, + now: () => now, + ttlMs: 10_000, }); - const idempotencyDir = path.join(rootDir, "idempotency"); - const pointerFiles = await fs.readdir(idempotencyDir); - expect(pointerFiles).toHaveLength(1); - const pointerFile = pointerFiles[0]; + const archive = Buffer.from("abc"); + const committed = await store.begin({ + kind: "skill-archive", + slug: "bounded-lease-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: committed.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: committed.uploadId }); - // corrupt the upload metadata so readRecordIfPresent() returns null on re-begin - // (this also drops the upload from the active-upload count) - await fs.rm(path.join(rootDir, first.uploadId, "metadata.json"), { force: true }); + const entered = deferred(); + const release = deferred(); + const pinned = store.withCommittedUpload(committed.uploadId, async () => { + entered.resolve(); + await release.promise; + }); + await entered.promise; + expect( + stateDatabase(databasePath) + .prepare( + "SELECT expires_at FROM state_leases WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .get(committed.uploadId), + ).toMatchObject({ expires_at: 1100 }); - // saturate the active-upload cap with unrelated uploads - for (let i = 0; i < ACTIVE_UPLOAD_LIMIT; i++) { - await store.begin({ kind: "skill-archive", slug: `filler-${i}`, sizeBytes: 8 }); - } + now = 1001; + release.resolve(); + await pinned; + }); - // re-begin the same key: the pointer still matches, but its upload metadata is gone, - // so the corrupt-metadata branch runs and then the active-upload cap throws before the - // pointer would be rewritten. The pointer must have been cleared by that branch — not - // left stranded pointing at the now-deleted (ghost) uploadId. - await expectUploadError( - store.begin({ kind: "skill-archive", slug: "demo-skill", sizeBytes: 8, idempotencyKey }), - "too many active skill uploads", - ); + it("does not renew an expired install lease", async () => { + const { databasePath, store } = await makeStore({ installLeaseHeartbeatMs: 60_000 }); + const archive = Buffer.from("abc"); + const committed = await store.begin({ + kind: "skill-archive", + slug: "expired-heartbeat-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: committed.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: committed.uploadId }); - const remaining = await fs.readdir(idempotencyDir).catch(() => [] as string[]); - expect(remaining).not.toContain(pointerFile); + await store.withCommittedUpload(committed.uploadId, async () => { + const db = stateDatabase(databasePath); + const lease = db + .prepare( + "SELECT owner FROM state_leases WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .get(committed.uploadId) as { owner: string }; + const heartbeatAt = Date.now(); + db.prepare( + "UPDATE state_leases SET expires_at = ? WHERE scope = 'skill-upload-install' AND lease_key = ?", + ).run(heartbeatAt - 1, committed.uploadId); + expect( + renewSkillUploadInstallLease({ + uploadId: committed.uploadId, + owner: lease.owner, + heartbeatAt, + expiresAt: heartbeatAt + 60_000, + options: { path: databasePath }, + }), + ).toBe(false); + }); + }); + + it("lets the install lease owner remove the upload", async () => { + const { databasePath, store } = await makeStore(); + const archive = Buffer.from("abc"); + const committed = await store.begin({ + kind: "skill-archive", + slug: "consumed-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: committed.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: committed.uploadId }); + let archivePath = ""; + await store.withCommittedUpload(committed.uploadId, async (record, controls) => { + archivePath = record.archivePath; + await controls.remove(); + }); + await expectMissingPath(archivePath); + expect(uploadCount(databasePath)).toBe(0); + expect(installLeaseCount(databasePath, committed.uploadId)).toBe(0); + }); + + it("does not remove an upload after the callback loses lease ownership", async () => { + const { databasePath, store } = await makeStore(); + const archive = Buffer.from("abc"); + const committed = await store.begin({ + kind: "skill-archive", + slug: "replacement-owner-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: committed.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: committed.uploadId }); + + await store.withCommittedUpload(committed.uploadId, async (_record, controls) => { + const replacementAt = Date.now(); + stateDatabase(databasePath) + .prepare( + "UPDATE state_leases SET owner = ?, expires_at = ?, updated_at = ? WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .run("replacement-owner", replacementAt + 60_000, replacementAt, committed.uploadId); + await expectUploadError(controls.remove(), "upload install lease is no longer active"); + }); + + expect(uploadCount(databasePath)).toBe(1); + expect( + stateDatabase(databasePath) + .prepare( + "SELECT owner FROM state_leases WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .get(committed.uploadId), + ).toMatchObject({ owner: "replacement-owner" }); + }); + + it("does not remove an upload after its install lease expires", async () => { + const { databasePath, store } = await makeStore(); + const archive = Buffer.from("abc"); + const committed = await store.begin({ + kind: "skill-archive", + slug: "expired-owner-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: committed.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: committed.uploadId }); + + await store.withCommittedUpload(committed.uploadId, async (_record, controls) => { + stateDatabase(databasePath) + .prepare( + "UPDATE state_leases SET expires_at = ? WHERE scope = 'skill-upload-install' AND lease_key = ?", + ) + .run(Date.now() - 1, committed.uploadId); + await expectUploadError(controls.remove(), "upload install lease is no longer active"); + }); + + expect(uploadCount(databasePath)).toBe(1); + }); + + it("cleans temporary materialization and preserves the upload on action failure", async () => { + const { databasePath, store } = await makeStore(); + const archive = Buffer.from("abc"); + + const committed = await store.begin({ + kind: "skill-archive", + slug: "throwing-skill", + sizeBytes: archive.length, + }); + await store.chunk({ + uploadId: committed.uploadId, + offset: 0, + dataBase64: archive.toString("base64"), + }); + await store.commit({ uploadId: committed.uploadId }); + let archivePath = ""; + await expect( + store.withCommittedUpload(committed.uploadId, async (record) => { + archivePath = record.archivePath; + throw new Error("action failed"); + }), + ).rejects.toThrow("action failed"); + await expectMissingPath(archivePath); + expect(uploadCount(databasePath)).toBe(1); }); }); diff --git a/src/skills/lifecycle/upload-store.ts b/src/skills/lifecycle/upload-store.ts index e673ba177a65..b8c549165505 100644 --- a/src/skills/lifecycle/upload-store.ts +++ b/src/skills/lifecycle/upload-store.ts @@ -7,15 +7,42 @@ import { isFutureDateTimestampMs, resolveExpiresAtMsFromDurationMs, } from "@openclaw/normalization-core/number-coercion"; -import { resolveStateDir } from "../../config/paths.js"; import { DEFAULT_MAX_ARCHIVE_BYTES_ZIP } from "../../infra/archive.js"; -import { sha256File, sha256Hex } from "../../infra/crypto-digest.js"; +import { sha256Hex } from "../../infra/crypto-digest.js"; import { formatErrorMessage } from "../../infra/errors.js"; -import { createAsyncLock, readDurableJsonFile, writeJsonAtomic } from "../../infra/json-files.js"; +import { createAsyncLock } from "../../infra/json-files.js"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "../../infra/kysely-sync.js"; +import { withTempWorkspace } from "../../infra/private-temp-workspace.js"; +import { resolvePreferredOpenClawTmpDir } from "../../infra/tmp-openclaw-dir.js"; +import { + openOpenClawStateDatabase, + runOpenClawStateWriteTransaction, + type OpenClawStateDatabaseOptions, +} from "../../state/openclaw-state-db.js"; import { validateRequestedSkillSlug } from "./archive-install.js"; +import { + deleteOwnedSkillUpload, + deleteSkillUploadState, + deleteExpiredSkillUploadUnlessLeased, + hasLiveSkillUploadInstallLease, + openSkillUploadDatabase, + readSkillUploadArchiveChunks, + readSkillUploadRow, + renewSkillUploadInstallLease, + resolveSkillUploadDatabaseOptions, + SKILL_UPLOAD_LEASE_SCOPE, + type SkillUploadDatabase, + type SkillUploadRow, +} from "./upload-store.sqlite.js"; /** Time window in which uploaded skill archive chunks may be committed. */ const SKILL_UPLOAD_TTL_MS = 60 * 60 * 1000; +const SKILL_UPLOAD_INSTALL_LEASE_MS = 15 * 60 * 1000; +const SKILL_UPLOAD_INSTALL_HEARTBEAT_MS = 30 * 1000; const MAX_SKILL_UPLOAD_CHUNK_BYTES = 4 * 1024 * 1024; const MAX_SKILL_UPLOAD_BASE64_LENGTH = Math.ceil(MAX_SKILL_UPLOAD_CHUNK_BYTES / 3) * 4; const MAX_ACTIVE_SKILL_UPLOADS = 32; @@ -24,7 +51,15 @@ const SKILL_UPLOAD_IDEMPOTENCY_KEY_MAX_LENGTH = 2048; const SHA256_PATTERN = /^[a-f0-9]{64}$/i; const UPLOAD_ID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; -const BASE64_PATTERN = /^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/; + +type SkillUploadStoreOptions = OpenClawStateDatabaseOptions & { + installLeaseHeartbeatMs?: number; + installLeaseMs?: number; + now?: () => number; + tempRootDir?: string; + ttlMs?: number; +}; + const locks = new Map; references: number }>(); export class SkillUploadRequestError extends Error { @@ -74,17 +109,6 @@ type CommitParams = { sha256?: string; }; -type IdempotencyRecord = { - version: 1; - keyHash: string; - uploadId: string; - kind: "skill-archive"; - slug: string; - force: boolean; - sizeBytes: number; - sha256?: string; -}; - async function withLock(key: string, fn: () => Promise): Promise { let entry = locks.get(key); if (!entry) { @@ -121,10 +145,6 @@ function validateUploadId(uploadId: string): string { return normalized; } -function isUploadId(value: string): boolean { - return UPLOAD_ID_PATTERN.test(value); -} - function validateSizeBytes(sizeBytes: number): number { if (!Number.isSafeInteger(sizeBytes) || sizeBytes < 1) { throw new SkillUploadRequestError("invalid sizeBytes"); @@ -161,45 +181,31 @@ function validateIdempotencyKey(value: string | undefined): string | undefined { return normalized; } -function hashText(value: string): string { - return sha256Hex(value); -} - -function resolveUploadsRoot(rootDir?: string): string { - return path.resolve(rootDir ?? path.join(resolveStateDir(), "tmp", "skill-uploads")); -} - -function resolveUploadDir(rootDir: string, uploadId: string): string { - return path.join(rootDir, validateUploadId(uploadId)); -} - -function resolveMetadataPath(rootDir: string, uploadId: string): string { - return path.join(resolveUploadDir(rootDir, uploadId), "metadata.json"); -} - -function resolveArchivePath(rootDir: string, uploadId: string): string { - return path.join(resolveUploadDir(rootDir, uploadId), "archive.zip"); -} - -function resolveIdempotencyPath(rootDir: string, keyHash: string): string { - return path.join(rootDir, "idempotency", `${keyHash}.json`); -} - -function estimateBase64DecodedBytes(value: string): number { - const padding = value.endsWith("==") ? 2 : value.endsWith("=") ? 1 : 0; - return (value.length / 4) * 3 - padding; +function resolvePositiveDuration(value: number | undefined, fallback: number): number { + return value !== undefined && Number.isSafeInteger(value) && value > 0 ? value : fallback; } function decodeBase64Chunk(dataBase64: string): Buffer { const normalized = dataBase64.trim(); - if (!normalized || normalized.length % 4 !== 0 || !BASE64_PATTERN.test(normalized)) { - throw new SkillUploadRequestError("invalid dataBase64"); - } if (normalized.length > MAX_SKILL_UPLOAD_BASE64_LENGTH) { throw new SkillUploadRequestError("upload chunk exceeds maximum size"); } - if (estimateBase64DecodedBytes(normalized) > MAX_SKILL_UPLOAD_CHUNK_BYTES) { - throw new SkillUploadRequestError("upload chunk exceeds maximum size"); + if (!normalized || normalized.length % 4 !== 0) { + throw new SkillUploadRequestError("invalid dataBase64"); + } + const paddingLength = normalized.endsWith("==") ? 2 : normalized.endsWith("=") ? 1 : 0; + const contentLength = normalized.length - paddingLength; + for (let index = 0; index < contentLength; index += 1) { + const code = normalized.charCodeAt(index); + const isBase64Character = + (code >= 0x41 && code <= 0x5a) || + (code >= 0x61 && code <= 0x7a) || + (code >= 0x30 && code <= 0x39) || + code === 0x2b || + code === 0x2f; + if (!isBase64Character) { + throw new SkillUploadRequestError("invalid dataBase64"); + } } const decoded = Buffer.from(normalized, "base64"); if (decoded.length < 1) { @@ -211,170 +217,166 @@ function decodeBase64Chunk(dataBase64: string): Buffer { return decoded; } -async function assertNotExpired( - rootDir: string, - record: SkillUploadRecord, - now: number, -): Promise { - const validNow = asDateTimestampMs(now); - if (validNow !== undefined && !isFutureDateTimestampMs(record.expiresAt, { nowMs: validNow })) { - await removeRecordFiles(rootDir, record); - throw new SkillUploadRequestError("upload has expired"); +function requireUploadRow(uploadId: string, options: OpenClawStateDatabaseOptions): SkillUploadRow { + const row = readSkillUploadRow(uploadId, options); + if (!row) { + throw new SkillUploadRequestError(`upload not found: ${uploadId}`); } + return row; +} + +function assertNotExpired( + row: SkillUploadRow, + nowMs: number, + options: OpenClawStateDatabaseOptions, +): void { + const validNow = asDateTimestampMs(nowMs); if (validNow === undefined) { throw new SkillUploadRequestError("upload has expired"); } + if (!isFutureDateTimestampMs(row.expires_at, { nowMs: validNow })) { + deleteExpiredSkillUploadUnlessLeased({ uploadId: row.upload_id, nowMs: validNow, options }); + throw new SkillUploadRequestError("upload has expired"); + } } -async function readRecord(rootDir: string, uploadId: string): Promise { - const record = await readDurableJsonFile( - resolveMetadataPath(rootDir, uploadId), +function matchesBegin( + row: SkillUploadRow, + params: { + kind: "skill-archive"; + slug: string; + force: boolean; + sizeBytes: number; + sha256?: string; + }, +): boolean { + return ( + row.kind === params.kind && + row.slug === params.slug && + row.force === (params.force ? 1 : 0) && + row.size_bytes === params.sizeBytes && + (row.sha256 ?? undefined) === params.sha256 ); - if (!record || record.version !== 1 || record.uploadId !== uploadId) { - throw new SkillUploadRequestError(`upload not found: ${uploadId}`); +} + +async function cleanupExpiredUploads(params: { + options: OpenClawStateDatabaseOptions; + nowMs: number; + lockRoot: string; + excludeUploadId?: string; +}): Promise { + const validNow = asDateTimestampMs(params.nowMs); + if (validNow === undefined) { + return; } - return { ...record, archivePath: resolveArchivePath(rootDir, uploadId) }; -} - -async function readRecordIfPresent( - rootDir: string, - uploadId: string, -): Promise { - const record = await readDurableJsonFile( - resolveMetadataPath(rootDir, uploadId), - ); - if (!record || record.version !== 1 || record.uploadId !== uploadId) { - return null; - } - return { - ...record, - archivePath: resolveArchivePath(rootDir, uploadId), - }; -} - -async function writeRecord(rootDir: string, record: SkillUploadRecord): Promise { - await writeJsonAtomic(resolveMetadataPath(rootDir, record.uploadId), record, { - mode: 0o600, - dirMode: 0o700, - trailingNewline: true, - }); -} - -async function removeUploadDir(rootDir: string, uploadId: string): Promise { - await fs.rm(resolveUploadDir(rootDir, uploadId), { recursive: true, force: true }); -} - -async function removeRecordFiles(rootDir: string, record: SkillUploadRecord): Promise { - await removeUploadDir(rootDir, record.uploadId); - if (record.idempotencyKeyHash) { - await fs.rm(resolveIdempotencyPath(rootDir, record.idempotencyKeyHash), { force: true }); - } -} - -async function listUploadIds(rootDir: string): Promise { - const entries = await fs.readdir(rootDir, { withFileTypes: true }).catch(() => []); - return entries - .filter((entry) => entry.isDirectory() && isUploadId(entry.name)) - .map((entry) => entry.name); -} - -async function cleanupExpiredUploads( - rootDir: string, - nowMs: number, - excludeUploadId?: string, -): Promise { - for (const uploadId of await listUploadIds(rootDir)) { - if (uploadId === excludeUploadId) { + const { database, kysely } = openSkillUploadDatabase(params.options); + const expired = executeSqliteQuerySync( + database.db, + kysely.selectFrom("skill_uploads").select("upload_id").where("expires_at", "<=", validNow), + ).rows; + for (const row of expired) { + if (row.upload_id === params.excludeUploadId) { continue; } - await withLock(`${rootDir}:upload:${uploadId}`, async () => { - const record = await readRecordIfPresent(rootDir, uploadId).catch(() => null); - const validNow = asDateTimestampMs(nowMs); - if ( - record && - validNow !== undefined && - !isFutureDateTimestampMs(record.expiresAt, { nowMs: validNow }) - ) { - await removeRecordFiles(rootDir, record); - } + await withLock(`${params.lockRoot}:upload:${row.upload_id}`, async () => { + runOpenClawStateWriteTransaction(({ db }) => { + const transactionDb = getNodeSqliteKysely(db); + if (hasLiveSkillUploadInstallLease(db, transactionDb, row.upload_id, validNow)) { + return; + } + const current = executeSqliteQueryTakeFirstSync( + db, + transactionDb + .selectFrom("skill_uploads") + .select("expires_at") + .where("upload_id", "=", row.upload_id), + ); + if (current && current.expires_at <= validNow) { + deleteSkillUploadState(db, transactionDb, row.upload_id); + } + }, params.options); }); } } -async function countActiveUploads(rootDir: string, nowMs: number): Promise { - let count = 0; - for (const uploadId of await listUploadIds(rootDir)) { - const record = await readRecordIfPresent(rootDir, uploadId).catch(() => null); - if (record && isFutureDateTimestampMs(record.expiresAt, { nowMs })) { - count += 1; +function assembleArchive( + chunks: Array<{ byte_offset: number; size_bytes: number; chunk_blob: Uint8Array }>, + expectedSize: number, +): Buffer { + let offset = 0; + const buffers: Buffer[] = []; + for (const chunk of chunks) { + const bytes = Buffer.from(chunk.chunk_blob); + if (chunk.byte_offset !== offset || chunk.size_bytes !== bytes.length || bytes.length < 1) { + throw new SkillUploadRequestError("uploaded archive chunks are incomplete"); } + buffers.push(bytes); + offset += bytes.length; } - return count; + if (offset !== expectedSize) { + throw new SkillUploadRequestError("uploaded archive chunks are incomplete"); + } + return Buffer.concat(buffers, expectedSize); } -async function writeArchiveChunk(params: { - archivePath: string; - offset: number; - decoded: Buffer; - afterSync: () => Promise; -}): Promise { - const handle = await fs.open(params.archivePath, "r+"); - try { - await handle.truncate(params.offset); - let written = 0; - while (written < params.decoded.length) { - const result = await handle.write( - params.decoded, - written, - params.decoded.length - written, - params.offset + written, - ); - if (result.bytesWritten <= 0) { - throw new Error("failed to write upload chunk"); - } - written += result.bytesWritten; - } - await handle.sync(); - await params.afterSync(); - } finally { - await handle.close().catch(() => undefined); - } +function toRecord(row: SkillUploadRow, archivePath: string): SkillUploadRecord { + return { + version: 1, + kind: "skill-archive", + uploadId: row.upload_id, + slug: row.slug, + force: row.force === 1, + sizeBytes: row.size_bytes, + ...(row.sha256 ? { sha256: row.sha256 } : {}), + ...(row.actual_sha256 ? { actualSha256: row.actual_sha256 } : {}), + receivedBytes: row.received_bytes, + archivePath, + createdAt: row.created_at, + expiresAt: row.expires_at, + committed: row.committed === 1, + ...(row.committed_at !== null ? { committedAt: row.committed_at } : {}), + ...(row.idempotency_key_hash ? { idempotencyKeyHash: row.idempotency_key_hash } : {}), + }; } -async function readCommittedRecord( - rootDir: string, - uploadId: string, - nowMs: number, -): Promise { - const record = await readRecord(rootDir, uploadId); - await assertNotExpired(rootDir, record, nowMs); - if (!record.committed) { - throw new SkillUploadRequestError("upload is not committed"); - } - if (!record.actualSha256) { +function toCommitResult(row: SkillUploadRow, requestedSha: string | undefined) { + if (!row.actual_sha256) { throw new SkillUploadRequestError("committed upload is missing sha256"); } - const stat = await fs.stat(record.archivePath).catch(() => null); - if (!stat || stat.size !== record.sizeBytes) { - throw new SkillUploadRequestError("uploaded archive is missing or incomplete"); + if (requestedSha && requestedSha !== row.actual_sha256) { + throw new SkillUploadRequestError("upload sha256 mismatch"); } - return record; + return { + uploadId: row.upload_id, + receivedBytes: row.received_bytes, + sha256: row.actual_sha256, + expiresAt: row.expires_at, + }; } -function createSkillUploadStore(options?: { - rootDir?: string; - now?: () => number; - ttlMs?: number; -}) { - const rootDir = resolveUploadsRoot(options?.rootDir); +function createSkillUploadStore(options?: SkillUploadStoreOptions) { + const stateOptions = resolveSkillUploadDatabaseOptions(options ?? {}); const now = options?.now ?? Date.now; const ttlMs = options?.ttlMs ?? SKILL_UPLOAD_TTL_MS; + const tempRootDir = options?.tempRootDir; + const installLeaseMs = resolvePositiveDuration( + options?.installLeaseMs, + SKILL_UPLOAD_INSTALL_LEASE_MS, + ); + const installLeaseHeartbeatMs = resolvePositiveDuration( + options?.installLeaseHeartbeatMs, + SKILL_UPLOAD_INSTALL_HEARTBEAT_MS, + ); + + function lockRoot(): string { + return openOpenClawStateDatabase(stateOptions).path; + } return { - rootDir, async begin(params: BeginParams) { - return await withLock(`${rootDir}:begin`, async () => { - await cleanupExpiredUploads(rootDir, now()); + const root = lockRoot(); + return await withLock(`${root}:begin`, async () => { + await cleanupExpiredUploads({ options: stateOptions, nowMs: now(), lockRoot: root }); if (params.kind !== "skill-archive") { throw new SkillUploadRequestError("unsupported upload kind"); } @@ -383,224 +385,352 @@ function createSkillUploadStore(options?: { const sha256 = normalizeSkillUploadSha256(params.sha256); const force = params.force === true; const idempotencyKey = validateIdempotencyKey(params.idempotencyKey); - const keyHash = idempotencyKey ? hashText(idempotencyKey) : undefined; - if (keyHash) { - const existing = await readDurableJsonFile( - resolveIdempotencyPath(rootDir, keyHash), - ); - if (existing) { - if ( - existing.kind !== params.kind || - existing.slug !== slug || - existing.force !== force || - existing.sizeBytes !== sizeBytes || - existing.sha256 !== sha256 - ) { - throw new SkillUploadRequestError("idempotencyKey conflicts with a different upload"); - } - const existingUploadId = validateUploadId(existing.uploadId); - const activeExisting = await withLock( - `${rootDir}:upload:${existingUploadId}`, - async () => { - const record = await readRecordIfPresent(rootDir, existingUploadId); - if (record && isFutureDateTimestampMs(record.expiresAt, { nowMs: now() })) { - return { - uploadId: record.uploadId, - receivedBytes: record.receivedBytes, - expiresAt: record.expiresAt, - }; - } - if (record) { - await removeRecordFiles(rootDir, record); - } else { - // Mirror removeRecordFiles for the corrupt/missing-metadata branch. - // The idempotency pointer still references this now-deleted upload, - // so drop it too. Otherwise, if the active-upload cap throws below - // before the pointer is rewritten, it strands an orphan idempotency - // file pointing at a ghost uploadId. - await removeUploadDir(rootDir, existingUploadId); - await fs.rm(resolveIdempotencyPath(rootDir, keyHash), { force: true }); - } - return null; - }, - ); - if (activeExisting) { - return activeExisting; - } - } - } - - if ((await countActiveUploads(rootDir, now())) >= MAX_ACTIVE_SKILL_UPLOADS) { - throw new SkillUploadRequestError("too many active skill uploads"); - } - - const uploadId = randomUUID(); - const uploadDir = resolveUploadDir(rootDir, uploadId); - const archivePath = resolveArchivePath(rootDir, uploadId); + const keyHash = idempotencyKey ? sha256Hex(idempotencyKey) : undefined; + // Expiry begins after cleanup waits, not before a possibly long in-flight install. const createdAt = now(); const expiresAt = resolveExpiresAtMsFromDurationMs(ttlMs, { nowMs: createdAt }); if (expiresAt === undefined) { throw new SkillUploadRequestError("invalid upload expiry"); } - const record: SkillUploadRecord = { - version: 1, - kind: params.kind, - uploadId, - slug, - force, - sizeBytes, - ...(sha256 ? { sha256 } : {}), - receivedBytes: 0, - archivePath, - createdAt, - expiresAt, - committed: false, - ...(keyHash ? { idempotencyKeyHash: keyHash } : {}), - }; - await fs.mkdir(uploadDir, { recursive: true, mode: 0o700 }); - await fs.writeFile(archivePath, Buffer.alloc(0), { mode: 0o600 }); - await writeRecord(rootDir, record); - if (keyHash) { - const idem: IdempotencyRecord = { - version: 1, - keyHash, - uploadId, - kind: params.kind, - slug, - force, - sizeBytes, - ...(sha256 ? { sha256 } : {}), - }; - await writeJsonAtomic(resolveIdempotencyPath(rootDir, keyHash), idem, { - mode: 0o600, - dirMode: 0o700, - trailingNewline: true, - }); - } - return { - uploadId, - receivedBytes: 0, - expiresAt: record.expiresAt, - }; + return runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + if (keyHash) { + const existing = executeSqliteQueryTakeFirstSync( + db, + kysely + .selectFrom("skill_uploads") + .selectAll() + .where("idempotency_key_hash", "=", keyHash), + ); + if (existing) { + if (!matchesBegin(existing, { kind: params.kind, slug, force, sizeBytes, sha256 })) { + throw new SkillUploadRequestError( + "idempotencyKey conflicts with a different upload", + ); + } + if (isFutureDateTimestampMs(existing.expires_at, { nowMs: createdAt })) { + return { + uploadId: existing.upload_id, + receivedBytes: existing.received_bytes, + expiresAt: existing.expires_at, + }; + } + if (hasLiveSkillUploadInstallLease(db, kysely, existing.upload_id, createdAt)) { + throw new SkillUploadRequestError("upload is already being installed"); + } + deleteSkillUploadState(db, kysely, existing.upload_id); + } + } + + const activeCount = executeSqliteQuerySync( + db, + kysely + .selectFrom("skill_uploads") + .select("upload_id") + .where("expires_at", ">", createdAt) + .limit(MAX_ACTIVE_SKILL_UPLOADS), + ).rows.length; + if (activeCount >= MAX_ACTIVE_SKILL_UPLOADS) { + throw new SkillUploadRequestError("too many active skill uploads"); + } + + const uploadId = randomUUID(); + executeSqliteQuerySync( + db, + kysely.insertInto("skill_uploads").values({ + upload_id: uploadId, + kind: params.kind, + slug, + force: force ? 1 : 0, + size_bytes: sizeBytes, + sha256: sha256 ?? null, + actual_sha256: null, + received_bytes: 0, + archive_blob: Buffer.alloc(0), + created_at: createdAt, + expires_at: expiresAt, + committed: 0, + committed_at: null, + idempotency_key_hash: keyHash ?? null, + }), + ); + return { uploadId, receivedBytes: 0, expiresAt }; + }, stateOptions); }); }, + async chunk(params: ChunkParams) { const uploadId = validateUploadId(params.uploadId); const offset = validateOffset(params.offset); const decoded = decodeBase64Chunk(params.dataBase64); - await cleanupExpiredUploads(rootDir, now(), uploadId); - return await withLock(`${rootDir}:upload:${uploadId}`, async () => { - const record = await readRecord(rootDir, uploadId); - await assertNotExpired(rootDir, record, now()); - if (record.committed) { - throw new SkillUploadRequestError("upload is already committed"); - } - if (offset !== record.receivedBytes) { - throw new SkillUploadRequestError( - `upload offset mismatch: expected ${record.receivedBytes}, got ${offset}`, + const root = lockRoot(); + await cleanupExpiredUploads({ + options: stateOptions, + nowMs: now(), + lockRoot: root, + excludeUploadId: uploadId, + }); + return await withLock(`${root}:upload:${uploadId}`, async () => { + const currentTime = now(); + assertNotExpired(requireUploadRow(uploadId, stateOptions), currentTime, stateOptions); + return runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + const row = executeSqliteQueryTakeFirstSync( + db, + kysely.selectFrom("skill_uploads").selectAll().where("upload_id", "=", uploadId), ); - } - const nextSize = record.receivedBytes + decoded.length; - if (nextSize > record.sizeBytes) { - throw new SkillUploadRequestError("upload chunk exceeds declared size"); - } - const nextRecord = { - ...record, - receivedBytes: nextSize, - }; - await writeArchiveChunk({ - archivePath: record.archivePath, - offset: record.receivedBytes, - decoded, - afterSync: async () => { - await writeRecord(rootDir, nextRecord); - }, - }); - return { - uploadId, - receivedBytes: nextRecord.receivedBytes, - expiresAt: nextRecord.expiresAt, - }; + if (!row) { + throw new SkillUploadRequestError(`upload not found: ${uploadId}`); + } + const validNow = asDateTimestampMs(currentTime); + if ( + validNow === undefined || + !isFutureDateTimestampMs(row.expires_at, { nowMs: validNow }) + ) { + throw new SkillUploadRequestError("upload has expired"); + } + if (row.committed === 1) { + throw new SkillUploadRequestError("upload is already committed"); + } + if (offset !== row.received_bytes) { + throw new SkillUploadRequestError( + `upload offset mismatch: expected ${row.received_bytes}, got ${offset}`, + ); + } + const nextSize = row.received_bytes + decoded.length; + if (nextSize > row.size_bytes) { + throw new SkillUploadRequestError("upload chunk exceeds declared size"); + } + executeSqliteQuerySync( + db, + kysely.insertInto("skill_upload_chunks").values({ + upload_id: uploadId, + byte_offset: offset, + size_bytes: decoded.length, + chunk_blob: decoded, + }), + ); + executeSqliteQuerySync( + db, + kysely + .updateTable("skill_uploads") + .set({ received_bytes: nextSize }) + .where("upload_id", "=", uploadId), + ); + return { uploadId, receivedBytes: nextSize, expiresAt: row.expires_at }; + }, stateOptions); }); }, + async commit(params: CommitParams) { const uploadId = validateUploadId(params.uploadId); const requestedSha = normalizeSkillUploadSha256(params.sha256); - return await withLock(`${rootDir}:upload:${uploadId}`, async () => { - const record = await readRecord(rootDir, uploadId); - await assertNotExpired(rootDir, record, now()); - if (record.committed) { - if (!record.actualSha256) { - throw new SkillUploadRequestError("committed upload is missing sha256"); - } - if (requestedSha && requestedSha !== record.actualSha256) { - throw new SkillUploadRequestError("upload sha256 mismatch"); - } - return { - uploadId, - receivedBytes: record.receivedBytes, - sha256: record.actualSha256, - expiresAt: record.expiresAt, - }; + const root = lockRoot(); + return await withLock(`${root}:upload:${uploadId}`, async () => { + const row = requireUploadRow(uploadId, stateOptions); + assertNotExpired(row, now(), stateOptions); + if (row.committed === 1) { + return toCommitResult(row, requestedSha); } - if (record.receivedBytes !== record.sizeBytes) { + if (row.received_bytes !== row.size_bytes) { throw new SkillUploadRequestError( - `upload size mismatch: expected ${record.sizeBytes}, got ${record.receivedBytes}`, + `upload size mismatch: expected ${row.size_bytes}, got ${row.received_bytes}`, ); } - const stat = await fs.stat(record.archivePath).catch(() => null); - if (!stat || stat.size !== record.sizeBytes) { - throw new SkillUploadRequestError("uploaded archive is missing or incomplete"); - } - if (record.sha256 && requestedSha && record.sha256 !== requestedSha) { + if (row.sha256 && requestedSha && row.sha256 !== requestedSha) { throw new SkillUploadRequestError("upload sha256 does not match begin sha256"); } - const actualSha256 = await sha256File(record.archivePath); - const expectedSha = requestedSha ?? record.sha256; + + let archive: Buffer; + try { + archive = assembleArchive( + readSkillUploadArchiveChunks(uploadId, stateOptions), + row.size_bytes, + ); + } catch (err) { + // Another process may commit and delete chunks after our metadata read. + // The committed parent row is the idempotent authority in that race. + const current = requireUploadRow(uploadId, stateOptions); + if (current.committed === 1) { + return toCommitResult(current, requestedSha); + } + throw err; + } + const actualSha256 = sha256Hex(archive); + const expectedSha = requestedSha ?? row.sha256 ?? undefined; if (expectedSha && expectedSha !== actualSha256) { throw new SkillUploadRequestError("upload sha256 mismatch"); } - const nextRecord = { - ...record, - sha256: record.sha256 ?? requestedSha ?? actualSha256, - actualSha256, - committed: true, - committedAt: now(), - }; - await writeRecord(rootDir, nextRecord); - return { - uploadId, - receivedBytes: nextRecord.receivedBytes, - sha256: actualSha256, - expiresAt: nextRecord.expiresAt, - }; + const committedAt = now(); + assertNotExpired(requireUploadRow(uploadId, stateOptions), committedAt, stateOptions); + + return runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + const current = executeSqliteQueryTakeFirstSync( + db, + kysely.selectFrom("skill_uploads").selectAll().where("upload_id", "=", uploadId), + ); + if (!current) { + throw new SkillUploadRequestError(`upload not found: ${uploadId}`); + } + if (current.committed === 1) { + return toCommitResult(current, requestedSha); + } + if (!isFutureDateTimestampMs(current.expires_at, { nowMs: committedAt })) { + throw new SkillUploadRequestError("upload has expired"); + } + if ( + current.received_bytes !== current.size_bytes || + current.size_bytes !== archive.length + ) { + throw new SkillUploadRequestError("uploaded archive chunks changed during commit"); + } + executeSqliteQuerySync( + db, + kysely + .updateTable("skill_uploads") + .set({ + actual_sha256: actualSha256, + archive_blob: archive, + committed: 1, + committed_at: committedAt, + }) + .where("upload_id", "=", uploadId), + ); + executeSqliteQuerySync( + db, + kysely.deleteFrom("skill_upload_chunks").where("upload_id", "=", uploadId), + ); + return { + uploadId, + receivedBytes: current.received_bytes, + sha256: actualSha256, + expiresAt: current.expires_at, + }; + }, stateOptions); }); }, + async withCommittedUpload( uploadIdRaw: string, action: (record: SkillUploadRecord, controls: { remove: () => Promise }) => Promise, ): Promise { const uploadId = validateUploadId(uploadIdRaw); - return await withLock(`${rootDir}:upload:${uploadId}`, async () => { - const record = await readCommittedRecord(rootDir, uploadId, now()); - return await action(record, { - remove: async () => { - await removeRecordFiles(rootDir, record); - }, - }); - }); - }, - async remove(uploadIdRaw: string): Promise { - const uploadId = validateUploadId(uploadIdRaw); - await withLock(`${rootDir}:upload:${uploadId}`, async () => { - const record = await readDurableJsonFile( - resolveMetadataPath(rootDir, uploadId), - ); - if (record && record.version === 1 && record.uploadId === uploadId) { - await removeRecordFiles(rootDir, record); - } else { - await removeUploadDir(rootDir, uploadId); + const root = lockRoot(); + return await withLock(`${root}:upload:${uploadId}`, async () => { + const leaseOwner = randomUUID(); + const currentTime = now(); + assertNotExpired(requireUploadRow(uploadId, stateOptions), currentTime, stateOptions); + const row = runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + const current = executeSqliteQueryTakeFirstSync( + db, + kysely.selectFrom("skill_uploads").selectAll().where("upload_id", "=", uploadId), + ); + if (!current) { + throw new SkillUploadRequestError(`upload not found: ${uploadId}`); + } + const validNow = asDateTimestampMs(currentTime); + if ( + validNow === undefined || + !isFutureDateTimestampMs(current.expires_at, { nowMs: validNow }) + ) { + throw new SkillUploadRequestError("upload has expired"); + } + if (current.committed !== 1) { + throw new SkillUploadRequestError("upload is not committed"); + } + if (!current.actual_sha256) { + throw new SkillUploadRequestError("committed upload is missing sha256"); + } + if (Buffer.from(current.archive_blob).length !== current.size_bytes) { + throw new SkillUploadRequestError("uploaded archive is missing or incomplete"); + } + executeSqliteQuerySync( + db, + kysely + .deleteFrom("state_leases") + .where("scope", "=", SKILL_UPLOAD_LEASE_SCOPE) + .where("lease_key", "=", uploadId) + .where("expires_at", "<=", currentTime), + ); + const claimed = executeSqliteQuerySync( + db, + kysely + .insertInto("state_leases") + .values({ + scope: SKILL_UPLOAD_LEASE_SCOPE, + lease_key: uploadId, + owner: leaseOwner, + expires_at: currentTime + installLeaseMs, + heartbeat_at: currentTime, + payload_json: null, + created_at: currentTime, + updated_at: currentTime, + }) + .onConflict((conflict) => conflict.doNothing()), + ); + if (claimed.numAffectedRows !== 1n) { + throw new SkillUploadRequestError("upload is already being installed"); + } + return current; + }, stateOptions); + + // The callback can cross process-local lock lifetimes. Keep the SQLite owner lease + // alive until it settles so another Gateway cannot sweep or reinstall this upload. + const heartbeat = setInterval(() => { + const heartbeatAt = now(); + try { + renewSkillUploadInstallLease({ + uploadId, + owner: leaseOwner, + heartbeatAt, + expiresAt: heartbeatAt + installLeaseMs, + options: stateOptions, + }); + } catch { + // A transient busy/error keeps the existing generous lease; the next tick retries. + } + }, installLeaseHeartbeatMs); + heartbeat.unref(); + + try { + return await withTempWorkspace( + { + rootDir: tempRootDir ?? resolvePreferredOpenClawTmpDir(), + prefix: "openclaw-skill-upload-", + }, + async (tmp) => { + const archivePath = path.join(tmp.dir, "archive.zip"); + await fs.writeFile(archivePath, Buffer.from(row.archive_blob), { mode: 0o600 }); + return await action(toRecord(row, archivePath), { + remove: async () => { + // Only the callback that still owns the install lease may consume the upload. + // A stalled callback must not erase a successor Gateway's replacement lease. + if ( + deleteOwnedSkillUpload(uploadId, leaseOwner, now(), stateOptions) === + "not-owner" + ) { + throw new SkillUploadRequestError("upload install lease is no longer active"); + } + }, + }); + }, + ); + } finally { + clearInterval(heartbeat); + runOpenClawStateWriteTransaction(({ db }) => { + const kysely = getNodeSqliteKysely(db); + executeSqliteQuerySync( + db, + kysely + .deleteFrom("state_leases") + .where("scope", "=", SKILL_UPLOAD_LEASE_SCOPE) + .where("lease_key", "=", uploadId) + .where("owner", "=", leaseOwner), + ); + }, stateOptions); } }); }, diff --git a/src/state/openclaw-state-db.generated.d.ts b/src/state/openclaw-state-db.generated.d.ts index 80be7ab81723..2c4a3472cbf9 100644 --- a/src/state/openclaw-state-db.generated.d.ts +++ b/src/state/openclaw-state-db.generated.d.ts @@ -897,6 +897,13 @@ export interface SkillLifecycle { state_changed_at_ms: number; } +export interface SkillUploadChunks { + byte_offset: number; + chunk_blob: Uint8Array; + size_bytes: number; + upload_id: string; +} + export interface SkillUploads { actual_sha256: string | null; archive_blob: Uint8Array; @@ -1273,6 +1280,7 @@ export interface DB { session_watch_cursors: SessionWatchCursors; skill_curator_state: SkillCuratorState; skill_lifecycle: SkillLifecycle; + skill_upload_chunks: SkillUploadChunks; skill_uploads: SkillUploads; skill_usage: SkillUsage; state_leases: StateLeases; diff --git a/src/state/openclaw-state-schema.generated.ts b/src/state/openclaw-state-schema.generated.ts index d3e84ebfb051..b8fb7650bf24 100644 --- a/src/state/openclaw-state-schema.generated.ts +++ b/src/state/openclaw-state-schema.generated.ts @@ -1010,6 +1010,16 @@ CREATE INDEX IF NOT EXISTS idx_skill_uploads_idempotency ON skill_uploads(idempotency_key_hash) WHERE idempotency_key_hash IS NOT NULL; +CREATE TABLE IF NOT EXISTS skill_upload_chunks ( + upload_id TEXT NOT NULL, + byte_offset INTEGER NOT NULL CHECK (byte_offset >= 0), + size_bytes INTEGER NOT NULL CHECK (size_bytes > 0), + chunk_blob BLOB NOT NULL, + PRIMARY KEY (upload_id, byte_offset), + FOREIGN KEY (upload_id) REFERENCES skill_uploads(upload_id) ON DELETE CASCADE, + CHECK (length(chunk_blob) = size_bytes) +); + CREATE TABLE IF NOT EXISTS capture_sessions ( id TEXT NOT NULL PRIMARY KEY, started_at INTEGER NOT NULL, diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index ee08d96ad019..f479379e5073 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -1005,6 +1005,16 @@ CREATE INDEX IF NOT EXISTS idx_skill_uploads_idempotency ON skill_uploads(idempotency_key_hash) WHERE idempotency_key_hash IS NOT NULL; +CREATE TABLE IF NOT EXISTS skill_upload_chunks ( + upload_id TEXT NOT NULL, + byte_offset INTEGER NOT NULL CHECK (byte_offset >= 0), + size_bytes INTEGER NOT NULL CHECK (size_bytes > 0), + chunk_blob BLOB NOT NULL, + PRIMARY KEY (upload_id, byte_offset), + FOREIGN KEY (upload_id) REFERENCES skill_uploads(upload_id) ON DELETE CASCADE, + CHECK (length(chunk_blob) = size_bytes) +); + CREATE TABLE IF NOT EXISTS capture_sessions ( id TEXT NOT NULL PRIMARY KEY, started_at INTEGER NOT NULL, diff --git a/test/scripts/check-database-first-legacy-stores.test.ts b/test/scripts/check-database-first-legacy-stores.test.ts index 1be1ca6af835..e55d4f1eb735 100644 --- a/test/scripts/check-database-first-legacy-stores.test.ts +++ b/test/scripts/check-database-first-legacy-stores.test.ts @@ -214,6 +214,19 @@ describe("check-database-first-legacy-stores", () => { expect(violations).toEqual([{ kind: "legacy store filesystem write", line: 4 }]); }); + it("flags runtime writes to retired skill-upload staging", () => { + const violations = collectDatabaseFirstLegacyStoreViolations( + ` + import { promises as fs } from "node:fs"; + import path from "node:path"; + await fs.writeFile(path.join(stateDir, "tmp", "skill-uploads", uploadId, "metadata.json"), "{}\n"); + `, + "src/skills/lifecycle/upload-file-store.ts", + ); + + expect(violations).toEqual([{ kind: "legacy store filesystem write", line: 4 }]); + }); + it("flags runtime writes to retired system-agent rescue approval stores", () => { const violations = collectDatabaseFirstLegacyStoreViolations( `