Prune async upload jobs and test polling flow

Author: snissn

packages/cli/src/index.ts

@@ -2589,6 +2589,8 @@ function startUiSiteServer(args: {
     lastErrorAt: null as string | null,
     lastSuccessAt: null as string | null
   };
+  const UPLOAD_JOB_TTL_MS = 10 * 60 * 1000;
+  const UPLOAD_JOB_MAX_COUNT = 128;
   const uploadJobs = new Map<
     string,
     {
@@ -2616,6 +2618,30 @@ function startUiSiteServer(args: {
   const faucetTargetEth = faucet?.targetWei ? Number(faucet.targetWei / 10n ** 18n) : 10;
   const manifestPaths = new Set(['/manifest.json', '/.well-known/tokenhost/manifest.json']);
 
+  function cleanupUploadJobs() {
+    const now = Date.now();
+    for (const [jobId, job] of uploadJobs.entries()) {
+      const updatedAt = Date.parse(String(job.updatedAt ?? job.createdAt ?? ''));
+      if (Number.isFinite(updatedAt) && now - updatedAt > UPLOAD_JOB_TTL_MS) {
+        uploadJobs.delete(jobId);
+      }
+    }
+
+    if (uploadJobs.size <= UPLOAD_JOB_MAX_COUNT) return;
+
+    const ordered = [...uploadJobs.entries()].sort((a, b) => {
+      const aTime = Date.parse(String(a[1].updatedAt ?? a[1].createdAt ?? '')) || 0;
+      const bTime = Date.parse(String(b[1].updatedAt ?? b[1].createdAt ?? '')) || 0;
+      return aTime - bTime;
+    });
+
+    while (ordered.length > UPLOAD_JOB_MAX_COUNT) {
+      const oldest = ordered.shift();
+      if (!oldest) break;
+      uploadJobs.delete(oldest[0]);
+    }
+  }
+
   function contentTypeForPath(filePath: string): string {
     const ext = path.extname(filePath).toLowerCase();
     switch (ext) {
@@ -2944,6 +2970,7 @@ function startUiSiteServer(args: {
 
     if (pathname === uploadPath || pathname === uploadStatusPath) {
       (async () => {
+        cleanupUploadJobs();
         const enabled = Boolean(upload?.enabled);
         const requestedUploadJobId = String(requestUrl.searchParams.get('jobId') ?? '').trim();
         if (req.method === 'GET' || req.method === 'HEAD') {

test/integration/testCliUploadPreviewIntegration.js

@@ -216,4 +216,156 @@ describe('CLI local preview upload integration', function () {
       preview.kill('SIGINT');
     }
   });
+
+  it('supports async foc-process uploads with job polling', async function () {
+    this.timeout(180000);
+
+    const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'th-upload-preview-foc-async-'));
+    const schemaPath = path.join(dir, 'schema.json');
+    const outDir = path.join(dir, 'out');
+    const fakeCliPath = path.join(dir, 'fake-foc-cli.mjs');
+    writeJson(schemaPath, uploadSchema());
+
+    fs.writeFileSync(
+      fakeCliPath,
+      `#!/usr/bin/env node
+import fs from 'fs';
+const args = process.argv.slice(2);
+const cmd = args[0];
+if (cmd === 'wallet' && args[1] === 'init') {
+  const keyIndex = args.indexOf('--privateKey');
+  const privateKey = keyIndex >= 0 ? String(args[keyIndex + 1] || '') : '';
+  const configDir = process.env.XDG_CONFIG_HOME
+    ? process.env.XDG_CONFIG_HOME + '/foc-cli-nodejs'
+    : process.env.HOME + '/.config/foc-cli-nodejs';
+  fs.mkdirSync(configDir, { recursive: true });
+  fs.writeFileSync(configDir + '/config.json', JSON.stringify({ privateKey }, null, 2));
+  process.stdout.write(JSON.stringify({ ok: true, data: { privateKey } }));
+  process.exit(0);
+}
+if (cmd === 'upload') {
+  const fileArg = args.find((value) => value && !value.startsWith('-') && value !== 'upload') || '';
+  const stat = fs.statSync(fileArg);
+  await new Promise((resolve) => setTimeout(resolve, 250));
+  process.stdout.write(JSON.stringify({
+    ok: true,
+    data: {
+      status: 'uploaded',
+      result: {
+        pieceCid: 'bafkfakeasyncuploadcid',
+        pieceScannerUrl: 'https://scanner.example/piece/bafkfakeasyncuploadcid',
+        size: stat.size,
+        copyResults: [
+          {
+            url: 'https://uploads.example.test/piece/bafkfakeasyncuploadcid',
+            providerRole: 'primary'
+          }
+        ],
+        copyFailures: []
+      },
+      processLog: [
+        { step: 'Reading file', status: 'done' },
+        { step: 'Uploading file', status: 'done' }
+      ]
+    }
+  }));
+  process.exit(0);
+}
+process.stderr.write(JSON.stringify({ ok: false, error: { message: 'unsupported fake foc command', args } }));
+process.exit(1);
+`
+    );
+    fs.chmodSync(fakeCliPath, 0o755);
+
+    const buildRes = runTh(['build', schemaPath, '--out', outDir], process.cwd(), {
+      TH_UPLOAD_RUNNER: 'foc-process',
+      TH_UPLOAD_PROVIDER: 'foc'
+    });
+    expect(buildRes.status, buildRes.stderr || buildRes.stdout).to.equal(0);
+
+    const port = 45200 + Math.floor(Math.random() * 1000);
+    const host = '127.0.0.1';
+    const baseUrl = `http://${host}:${port}`;
+
+    const preview = spawn(
+      'node',
+      [
+        path.resolve('packages/cli/dist/index.js'),
+        'preview',
+        outDir,
+        '--host',
+        host,
+        '--port',
+        String(port),
+        '--no-deploy',
+        '--no-start-anvil',
+        '--no-faucet'
+      ],
+      {
+        cwd: process.cwd(),
+        stdio: ['ignore', 'pipe', 'pipe'],
+        env: {
+          ...process.env,
+          TH_UPLOAD_RUNNER: 'foc-process',
+          TH_UPLOAD_PROVIDER: 'foc',
+          TH_UPLOAD_FOC_COMMAND: `node ${fakeCliPath}`,
+          TH_UPLOAD_FOC_DEBUG: '1',
+          TH_UPLOAD_FOC_COPIES: '1',
+          TH_UPLOAD_FOC_CHAIN: '314159',
+          PRIVATE_KEY: 'fff91c6963a11a8ff48f13297185f110678b47086992b0f1612b7a1467d11f0c',
+          XDG_CONFIG_HOME: path.join(dir, 'xdg-config')
+        }
+      }
+    );
+
+    try {
+      await waitForOutput(preview, new RegExp(`http://${host}:${port}/`), 60000);
+
+      const uploadStatus = await request(`${baseUrl}/__tokenhost/upload`);
+      expect(uploadStatus.status).to.equal(200);
+      expect(uploadStatus.json?.ok).to.equal(true);
+      expect(uploadStatus.json?.runnerMode).to.equal('foc-process');
+      expect(uploadStatus.json?.provider).to.equal('filecoin_onchain_cloud');
+
+      const payload = Buffer.from('fake-foc-async-upload', 'utf-8');
+      const accepted = await request(`${baseUrl}/__tokenhost/upload`, {
+        method: 'POST',
+        headers: {
+          'content-type': 'image/png',
+          'x-tokenhost-upload-filename': 'async.png',
+          'x-tokenhost-upload-size': String(payload.length),
+          'x-tokenhost-upload-mode': 'async'
+        },
+        body: payload
+      });
+
+      expect(accepted.status).to.equal(202);
+      expect(accepted.json?.ok).to.equal(true);
+      expect(accepted.json?.pending).to.equal(true);
+      expect(String(accepted.json?.jobId || '')).to.not.equal('');
+      expect(String(accepted.json?.statusUrl || '')).to.match(/^\/__tokenhost\/upload\?jobId=/);
+
+      let completed = null;
+      for (let attempt = 0; attempt < 20; attempt += 1) {
+        await new Promise((resolve) => setTimeout(resolve, 250));
+        const polled = await request(`${baseUrl}${accepted.json.statusUrl}`);
+        expect(polled.status).to.equal(200);
+        if (polled.json?.pending) continue;
+        completed = polled.json;
+        break;
+      }
+
+      expect(completed?.ok).to.equal(true);
+      expect(completed?.done).to.equal(true);
+      expect(completed?.upload?.url).to.equal('https://uploads.example.test/piece/bafkfakeasyncuploadcid');
+      expect(completed?.upload?.cid).to.equal('bafkfakeasyncuploadcid');
+
+      const finalStatus = await request(`${baseUrl}/__tokenhost/upload`);
+      expect(finalStatus.status).to.equal(200);
+      expect(finalStatus.json?.lastError).to.equal(null);
+      expect(String(finalStatus.json?.lastSuccessAt || '')).to.not.equal('');
+    } finally {
+      preview.kill('SIGINT');
+    }
+  });
 });