[pull] main from TryGhost:main

.github/actions/load-docker-image/action.yml

@@ -8,7 +8,7 @@ inputs:
     description: 'Docker image tags (multi-line string)'
     required: true
   artifact-name:
-    description: 'Name of the artifact to download (fork PRs only)'
+    description: 'Name of the image artifact to download'
     required: false
     default: 'docker-image'
 
@@ -20,15 +20,17 @@ runs:
       uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8
       with:
         name: ${{ inputs.artifact-name }}
+        path: ${{ runner.temp }}/${{ inputs.artifact-name }}
 
     - name: Load image from artifact (artifact)
       if: inputs.use-artifact == 'true'
       shell: bash
       env:
-        ARTIFACT_NAME: ${{ inputs.artifact-name }}
+        ARTIFACT_PATH: ${{ runner.temp }}/${{ inputs.artifact-name }}/${{ inputs.artifact-name }}.tar.gz
       run: |
+        trap 'rm -f -- "$ARTIFACT_PATH"' EXIT
         echo "Loading Docker image from artifact..."
-        gunzip -c "${ARTIFACT_NAME}.tar.gz" | docker load
+        gunzip -c "$ARTIFACT_PATH" | docker load
         echo "Available images after load:"
         docker images
 

.github/workflows/ci.yml

@@ -2060,28 +2060,33 @@ jobs:
       - name: Setup Docker Registry Mirrors
         uses: ./.github/actions/setup-docker-registry-mirrors
 
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4
+      - name: Pull Tinybird CLI image
+        id: pull_tb_cli
+        if: matrix.analytics == 'true' && needs.job_setup.outputs.changed_tb_cli != 'true'
+        run: |
+          COMPOSE_IMAGE="${COMPOSE_PROJECT_NAME:-ghost-dev}-tb-cli"
+          if docker pull ghcr.io/tryghost/tb-cli:latest; then
+            docker tag ghcr.io/tryghost/tb-cli:latest "$COMPOSE_IMAGE"
+            echo "Pulled tb-cli from GHCR"
+            echo "build_required=false" >> "$GITHUB_OUTPUT"
+          else
+            echo "GHCR image not available, building from source"
+            echo "build_required=true" >> "$GITHUB_OUTPUT"
+          fi
 
-      - name: Pull or build Tinybird CLI Image
-        if: matrix.analytics == 'true'
+      - name: Build Tinybird CLI image
+        if: matrix.analytics == 'true' && (needs.job_setup.outputs.changed_tb_cli == 'true' || steps.pull_tb_cli.outputs.build_required == 'true')
         env:
           CHANGED_TB_CLI: ${{ needs.job_setup.outputs.changed_tb_cli }}
         run: |
           COMPOSE_IMAGE="${COMPOSE_PROJECT_NAME:-ghost-dev}-tb-cli"
-          # GHCR's :latest is only republished from main, so it can't reflect a
-          # Dockerfile change under review — build from source when this PR
-          # touches it, otherwise take the prebuilt fast path.
           if [[ "$CHANGED_TB_CLI" == 'true' ]]; then
             echo "docker/tb-cli changed, building from source"
-            docker buildx build --load -t "$COMPOSE_IMAGE" -f docker/tb-cli/Dockerfile .
-          elif docker pull ghcr.io/tryghost/tb-cli:latest 2>/dev/null; then
-            echo "Pulled tb-cli from GHCR"
-            docker tag ghcr.io/tryghost/tb-cli:latest "$COMPOSE_IMAGE"
           else
             echo "GHCR image not available, building from source"
-            docker buildx build --load -t "$COMPOSE_IMAGE" -f docker/tb-cli/Dockerfile .
           fi
+          docker build --tag "$COMPOSE_IMAGE" --file docker/tb-cli/Dockerfile .
+          docker builder prune --all --force
 
       - name: Load Image
         uses: ./.github/actions/load-docker-image
@@ -2103,6 +2108,12 @@ jobs:
         # (admin, apps, ghost/core). Scope the install to cut shard setup time.
         run: pnpm install --frozen-lockfile --filter @tryghost/e2e...
 
+      - name: Report Analytics runner disk usage
+        if: matrix.analytics == 'true'
+        run: |
+          df -h /
+          docker system df
+
       - name: Prepare E2E CI job
         env:
           GHOST_E2E_IMAGE: ${{ steps.load.outputs.image-tag }}

ghost/core/core/server/api/endpoints/posts.js

@@ -129,9 +129,8 @@ const controller = {
       method: 'importContent',
     },
     async query(frame) {
-      // The CSV must be parsed before the response goes out: the uploaded temp file is
-      // deleted as soon as it is sent. The posts are written by a background job behind
-      // the 202.
+      // The content-import service stages and validates the upload before this response;
+      // Multer deletes the request temp file as soon as the 202 is sent.
       const { importId, total } = await contentImportService.importCSV({
         filePath: frame.file.path,
         fileName: frame.file.name,

ghost/core/core/server/services/content-import/import/importer.ts

@@ -13,16 +13,18 @@ import type { Clock, ImportRun, ImportRunStore, RowOutcome } from './store';
 import type { PreparedImportSource } from './source';
 import type { PreparedPostRow, PreparedPostRows } from './reader';
 import { MediaInliningFailure, type PostMediaInlining } from './media';
+import type { ImportFileStager, StagedImportFile } from './staged-file';
+import ContentCSVImportJob from '../jobs/content-csv-import-job';
 
 export type { ImportRequest } from './schema';
 
 const errors = require('@tryghost/errors');
 const logging = require('@tryghost/logging');
 const tpl = require('@tryghost/tpl');
 
-// The CSV is parsed inside the request (the uploaded temp file is deleted when the
-// response is sent); the parsed rows are handed to an in-process background job
-// that writes one post per row.
+// The upload is staged because the request temp file is deleted with the response.
+// Request-time parsing preserves synchronous validation and the current row count;
+// the class-based job reparses the staged file so its payload stays serializable.
 
 // The id is what a completion report will be looked up by.
 export interface ImportAccepted {
@@ -54,7 +56,7 @@ const messages = {
 
 function logLifecycle(message: string): void {
   try {
-    logging.info(`[Background Job] content-import ${message}`);
+    logging.info(`[Background Job] ${ContentCSVImportJob.type} ${message}`);
   } catch {
     // Observability must not change whether an import is queued or resolves.
   }
@@ -72,7 +74,8 @@ interface ImporterDeps {
   getCleanHTML: () => CleanHTML;
   createMediaInliner: () => PostMediaInlining;
   email: EmailNotifications;
-  addJob: (job: { job: () => Promise<void>; offloaded: boolean; name: string }) => void;
+  dispatchJob: (job: ContentCSVImportJob) => Promise<void>;
+  fileStager: ImportFileStager;
   report: FailureReporter;
   store: ImportRunStore;
   urlForPost: (post: WrittenPost) => string;
@@ -97,7 +100,8 @@ class ContentCSVImporter {
   private _getCleanHTML: () => CleanHTML;
   private _createMediaInliner: () => PostMediaInlining;
   private _email: EmailNotifications;
-  private _addJob: ImporterDeps['addJob'];
+  private _dispatchJob: ImporterDeps['dispatchJob'];
+  private _fileStager: ImportFileStager;
   private _report: FailureReporter;
   private _store: ImportRunStore;
   private _urlForPost: (post: WrittenPost) => string;
@@ -114,7 +118,8 @@ class ContentCSVImporter {
     getCleanHTML,
     createMediaInliner,
     email,
-    addJob,
+    dispatchJob,
+    fileStager,
     report,
     store,
     urlForPost,
@@ -130,7 +135,8 @@ class ContentCSVImporter {
     this._getCleanHTML = getCleanHTML;
     this._createMediaInliner = createMediaInliner;
     this._email = email;
-    this._addJob = addJob;
+    this._dispatchJob = dispatchJob;
+    this._fileStager = fileStager;
     this._report = report;
     this._store = store;
     this._urlForPost = urlForPost;
@@ -141,66 +147,103 @@ class ContentCSVImporter {
 
   async importCSV(request: ImportRequest): Promise<ImportAccepted> {
     const emailRecipient = request.requestUserEmail ?? (await this._email.getDefaultRecipient());
-    const source = await this._prepareSource(request);
-    let preparedRows: PreparedPostRows;
+    let stagedFile: StagedImportFile;
     try {
-      const result = await this._readRows(source.filePath, request.mapping);
-      preparedRows = Array.isArray(result)
-        ? { columns: [], rows: result.map((data, index) => ({ data, line: index + 2 })) }
-        : result;
+      stagedFile = await this._fileStager.stage(request);
     } catch (error) {
-      await this.cleanupSource(source.cleanup);
       throw new errors.ValidationError({
         message: tpl(messages.unreadableFile),
         err: error,
       });
     }
+    let handedOff = false;
 
-    // Temporary while import state is held in memory: the durable job
-    // system milestone removes the cap.
-    if (preparedRows.rows.length > MAX_POSTS) {
+    try {
+      const stagedRequest = {
+        filePath: stagedFile.path,
+        fileName: stagedFile.name,
+        mapping: request.mapping,
+        requestUserEmail: request.requestUserEmail,
+      };
+      const { source, preparedRows } = await this.readPreparedRows(stagedRequest);
       await this.cleanupSource(source.cleanup);
-      throw new errors.ValidationError({
-        message: tpl(messages.tooManyPosts, { max: MAX_POSTS }),
+
+      this.assertWithinPostLimit(preparedRows.rows);
+
+      const runId = this._newRunId();
+      const importTagNames = buildImportTagNames(runId, this._getTimezone(), this._now());
+      this._store.create(runId, preparedRows.rows.length, preparedRows.columns);
+      const job = new ContentCSVImportJob({
+        importId: runId,
+        file: stagedFile,
+        mapping: request.mapping,
+        importTagNames,
+        emailRecipient,
       });
+
+      logLifecycle('queued');
+      try {
+        await this._dispatchJob(job);
+      } catch (error) {
+        this._store.fail(runId, messageOf(error));
+        this._store.release(runId);
+        throw error;
+      }
+
+      handedOff = true;
+      return { importId: runId, total: preparedRows.rows.length };
+    } finally {
+      if (!handedOff) {
+        await this.cleanupStagedFile(stagedFile);
+      }
     }
+  }
 
-    const runId = this._newRunId();
-    const importTagNames = buildImportTagNames(runId, this._getTimezone(), this._now());
-    this._store.create(runId, preparedRows.rows.length, preparedRows.columns);
+  async handle(job: ContentCSVImportJob): Promise<void> {
+    let source: PreparedImportSource | undefined;
 
-    logLifecycle('queued');
     try {
-      this._addJob({
-        job: () =>
-          this.runImportJob(runId, importTagNames, preparedRows.rows, source, emailRecipient),
-        offloaded: false,
-        name: 'content-import',
+      const prepared = await this.readPreparedRows({
+        filePath: job.file.path,
+        fileName: job.file.name,
+        mapping: job.mapping,
       });
+      source = prepared.source;
+      this.assertWithinPostLimit(prepared.preparedRows.rows);
+      await this.processRows(
+        job.importId,
+        job.importTagNames,
+        prepared.preparedRows.rows,
+        prepared.source,
+      );
     } catch (error) {
-      this._store.fail(runId, messageOf(error));
-      await this.cleanupSource(source.cleanup);
-      this._store.release(runId);
+      this._store.fail(job.importId, messageOf(error));
       throw error;
+    } finally {
+      if (source) {
+        await this.cleanupSource(source.cleanup);
+      }
+      const run = this._store.get(job.importId);
+      if (run) {
+        await this.settle(() => this._email.send(run, job.emailRecipient));
+      }
+      this._store.release(job.importId);
+      await this.cleanupStagedFile(job.file);
     }
+  }
 
-    return { importId: runId, total: preparedRows.rows.length };
+  allSettled(): Promise<void> {
+    return this._store.allSettled();
   }
 
-  // Must resolve in every case: the job manager reads a rejected inline job as a
-  // defect in the job itself, and there is no retry behind it.
-  private async runImportJob(
+  private async processRows(
     runId: string,
     importTagNames: string[],
     rows: PreparedPostRow[],
     source: PreparedImportSource,
-    emailRecipient: string,
   ): Promise<void> {
-    const startedAt = Date.now();
-    logLifecycle('started');
     let urlFailureCount = 0;
     let firstUrlFailure: unknown;
-    let failed = false;
 
     try {
       if (source.assets) {
@@ -347,19 +390,36 @@ class ContentCSVImporter {
       this.reportUrlFailures(urlFailureCount, firstUrlFailure);
       this._store.finish(runId);
     } catch (error) {
-      failed = true;
       this.reportUrlFailures(urlFailureCount, firstUrlFailure);
-      this._report(error);
-      this._store.fail(runId, messageOf(error));
-    } finally {
+      throw error;
+    }
+  }
+
+  private async readPreparedRows(
+    request: ImportRequest,
+  ): Promise<{ source: PreparedImportSource; preparedRows: PreparedPostRows }> {
+    const source = await this._prepareSource(request);
+    try {
+      const result = await this._readRows(source.filePath, request.mapping);
+      const preparedRows = Array.isArray(result)
+        ? { columns: [], rows: result.map((data, index) => ({ data, line: index + 2 })) }
+        : result;
+      return { source, preparedRows };
+    } catch (error) {
       await this.cleanupSource(source.cleanup);
-      const run = this._store.get(runId);
-      if (run) {
-        await this.settle(() => this._email.send(run, emailRecipient));
-      }
-      this._store.release(runId);
-      const outcome = failed ? 'failed after' : 'completed in';
-      logLifecycle(`${outcome} ${Date.now() - startedAt}ms`);
+      throw new errors.ValidationError({
+        message: tpl(messages.unreadableFile),
+        err: error,
+      });
+    }
+  }
+
+  private assertWithinPostLimit(rows: PreparedPostRow[]): void {
+    // Temporary while import state is held in memory: M8 removes the cap.
+    if (rows.length > MAX_POSTS) {
+      throw new errors.ValidationError({
+        message: tpl(messages.tooManyPosts, { max: MAX_POSTS }),
+      });
     }
   }
 
@@ -393,6 +453,14 @@ class ContentCSVImporter {
     }
   }
 
+  private async cleanupStagedFile(file: StagedImportFile): Promise<void> {
+    try {
+      await this._fileStager.remove(file);
+    } catch (error) {
+      this._report(error);
+    }
+  }
+
   private async settle(operation: () => Promise<unknown>): Promise<void> {
     try {
       await operation();