vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-02 03:11:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(state): move skill upload staging to SQLite (#108346)

Browse Source 
* 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
This commit is contained in:
Peter Steinberger
2026-07-15 08:40:46 -07:00
committed by GitHub
parent 09ce4afd16
commit f41c143345
13 changed files with 1458 additions and 596 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
docs/refactor/database-first.md
View File

@@ -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)

2
scripts/check-database-first-legacy-stores.mjs
View File

@@ -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",

37
src/commands/doctor-usage-cost-cache.test.ts
View File

@@ -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";
}

36
src/commands/doctor-usage-cost-cache.ts
View File

@@ -89,10 +89,46 @@ async function maybeRemoveLegacyUsageCostCacheFiles(params: {
);
}
async function maybeRemoveLegacySkillUploadTree(params: {
shouldRepair: boolean;
env?: NodeJS.ProcessEnv;
homedir?: () => string;
}): Promise<void> {
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<void> {
await maybeScrubConfigAuditLog({ shouldRepair, env });
await maybeRemoveLegacyUsageCostCacheFiles({ shouldRepair, env });
await maybeRemoveLegacySkillUploadTree({ shouldRepair, env });
}

45
src/gateway/server-methods/skills-upload.test.ts
View File

@@ -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<void> {
}
}
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);
});
});

190
src/skills/lifecycle/upload-store.sqlite.ts Normal file
View File

@@ -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<SkillUploads>;
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<SkillUploadDatabase>(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<SkillUploadDatabase>,
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<SkillUploadDatabase>(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<SkillUploadDatabase>,
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<SkillUploadDatabase>(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<SkillUploadDatabase>(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;
}

6
src/skills/lifecycle/upload-store.test-support.ts
View File

@@ -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;
};

818
src/skills/lifecycle/upload-store.test.ts
View File

File diff suppressed because it is too large Load Diff

868
src/skills/lifecycle/upload-store.ts
View File

@@ -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<string, { lock: ReturnType<typeof createAsyncLock>; 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<T>(key: string, fn: () => Promise<T>): Promise<T> {
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<void> {
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<SkillUploadRecord> {
const record = await readDurableJsonFile<SkillUploadRecord>(
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<void> {
const validNow = asDateTimestampMs(params.nowMs);
if (validNow === undefined) {
return;
}
return { ...record, archivePath: resolveArchivePath(rootDir, uploadId) };
}
async function readRecordIfPresent(
rootDir: string,
uploadId: string,
): Promise<SkillUploadRecord | null> {
const record = await readDurableJsonFile<SkillUploadRecord>(
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<void> {
await writeJsonAtomic(resolveMetadataPath(rootDir, record.uploadId), record, {
mode: 0o600,
dirMode: 0o700,
trailingNewline: true,
});
}
async function removeUploadDir(rootDir: string, uploadId: string): Promise<void> {
await fs.rm(resolveUploadDir(rootDir, uploadId), { recursive: true, force: true });
}
async function removeRecordFiles(rootDir: string, record: SkillUploadRecord): Promise<void> {
await removeUploadDir(rootDir, record.uploadId);
if (record.idempotencyKeyHash) {
await fs.rm(resolveIdempotencyPath(rootDir, record.idempotencyKeyHash), { force: true });
}
}
async function listUploadIds(rootDir: string): Promise<string[]> {
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<void> {
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<SkillUploadDatabase>(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<number> {
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<void>;
}): Promise<void> {
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<SkillUploadRecord> {
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<IdempotencyRecord>(
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<SkillUploadDatabase>(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<SkillUploadDatabase>(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<SkillUploadDatabase>(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<T>(
uploadIdRaw: string,
action: (record: SkillUploadRecord, controls: { remove: () => Promise<void> }) => Promise<T>,
): Promise<T> {
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<void> {
const uploadId = validateUploadId(uploadIdRaw);
await withLock(`${rootDir}:upload:${uploadId}`, async () => {
const record = await readDurableJsonFile<SkillUploadRecord>(
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<SkillUploadDatabase>(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<SkillUploadDatabase>(db);
executeSqliteQuerySync(
db,
kysely
.deleteFrom("state_leases")
.where("scope", "=", SKILL_UPLOAD_LEASE_SCOPE)
.where("lease_key", "=", uploadId)
.where("owner", "=", leaseOwner),
);
}, stateOptions);
}
});
},

8
src/state/openclaw-state-db.generated.d.ts vendored
View File

@@ -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;

10
src/state/openclaw-state-schema.generated.ts
View File

@@ -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,

10
src/state/openclaw-state-schema.sql
View File

@@ -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,

13
test/scripts/check-database-first-legacy-stores.test.ts
View File

@@ -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(
`