Enterprise 版本

Stream：文件工作處理

概觀

NextPDF\Enterprise\Stream\DocumentJobStreamProcessor 將一連串 render manifest 轉為持久且可課責的結果。它以 generator 的形式消費 iterable<RenderManifest>，透過 Pro render engine 渲染有界的視窗，並依來源順序終結每一項工作。每項工作都會結束於恰好一種終態：輸出已提交、被辨識為早已提交，或被送入死信。進度會被檢查點記錄下來，因此當機的執行可以恢復而不會重新發布任何內容。

Stream 的故事橫跨兩個版本，而這樣的切分是刻意為之。Pro 提供持久且可並行的 render engine，以及本機、單一主機的檔案系統儲存——也就是行程內的那一半。Enterprise 則提供這個文件工作串流處理器，加上跨越主機邊界的元件：物件儲存提交器（ObjectStorageCommitter）與終態事件的持久 outbox（FilesystemOutboxEmitter）。Pro Stream 頁面從它那一側陳述了同一條邊界。

可用性與授權

此能力隨 NextPDF Enterprise（nextpdf/enterprise）出貨，並以 Enterprise 級授權封套啟用。沒有該權利的部署並不會載入此能力的類別。比較各版本並取得授權。

安裝

composer require nextpdf/enterprise

本頁的類別位於 NextPDF\Enterprise\Stream 與 NextPDF\Enterprise\Stream\Storage 之下。它們消費 NextPDF\Pro\Stream 中已凍結的 Pro 合約——engine、committer、checkpoint、idempotency、retry 與 dead-letter 介面。

概念總覽

處理器的職責是交付語意，而非渲染。它把 manifest 串流分組成不大於 engine 批次大小的來源偏移量視窗。每個視窗透過 RenderEngineInterface::renderBatch() 渲染，並對每項項目的逾時進行有界、確定性的重試。接著每一項項目都會依來源偏移量順序終結到一個終態結果。

exactly-once 的邊界是逐項目的，而它錨定在 committer，而非協調機制。持久的進度是一個以 1 為起點的偏移量高水位：位於檢查點或其之下的每個偏移量都已到達終態結果。屏障順序是固定的：提交位元組、推進高水位、儲存檢查點、沖刷緩衝的 idempotency 標記，然後才發出終態事件。介於提交與檢查點之間的當機會在恢復時以冪等方式重新提交，因為 committer 會比對摘要。檢查點之後的當機會快轉越過該偏移量，因此沒有任何東西會被發布兩次。

ObjectStorageCommitter 透過最小的 ObjectStorageClientInterface 針對物件儲存實作 Pro 的 OutputCommitterInterface。目標的 container 是 bucket，其 key 則是物件鍵。重新提交相同位元組是經摘要比對後的 no-op。在沒有 overwrite 的情況下提交相異位元組會引發 SPEC-COMMIT-409 衝突。全新物件只會透過原子的 putIfAbsent() 條件式寫入被建立；在該競賽中落敗會觸發一個有界的重讀並解決迴圈。因此跨寫入者的 exactly-once 只在你的轉接器 putIfAbsent() 是真正條件式寫入的範圍內成立——在 S3 上是 If-None-Match: *，在 GCS 上是 ifGenerationMatch: 0。本週期出貨的是介面加上記憶體內的 NullObjectStorageClient；實際的 S3/GCS 轉接器由主機端提供。

終態事件為下游系統收攏這個迴圈。在檢查點屏障之後，處理器會嘗試為每項已終結的工作發出一個 JobTerminalEvent——識別碼、狀態、收據、錯誤細節、嘗試次數，且絕不含任何 PDF 位元組。搭配單純的 callback emitter 時，發出是 at-most-once：檢查點之後的事件可能在當機恢復時被略過。FilesystemOutboxEmitter 讓每個事件在 emit() 執行後即成為持久：每個事件都是一個以其確定性 eventId 雜湊命名的原子 JSON 檔，因此恢復後重新發出是冪等的、一個轉送器會 at-least-once 交付、消費端則以 eventId 去重。無論哪種方式都留有一條邊界：發出發生在檢查點屏障之後，因此介於 checkpoint.save() 與 emit() 之間的當機會在恢復時略過該項目的終態事件。需要完整事件帳簿的下游系統，應對照已提交的物件（該儲存才是真實來源）進行對帳，而非僅對照 outbox。

授權被接進了輸出路徑。withBrandingFromLicense() 工廠會在每次執行時從授權解析一次評估版品牌策略。付費授權解析為恆等轉換。評估版或缺失的授權會在每份已提交文件上加浮水印，而無法被套上品牌的文件會被送入死信——處理器絕不會提交未加品牌的評估版位元組。

為何這樣運作

承載一切的決策是：exactly-once 建立在 committer 那經摘要比對、以條件方式建立的物件之上——而非分散式鎖或共識。物件儲存的條件式寫入是本設計唯一需要的原子基元，其餘一切都被允許失敗並恢復。這正是為何 render engine 必須維持無副作用、為何有鍵狀態與執行內去重快取被視為可重新計算的加速手段，也是為何一個含糊的提交會中止執行而非猜測：恢復路徑會透過同一次摘要比對收斂。這也是為何每個 runId 單一寫入者被定為明文要求，而非強制執行的租約——檢查點儲存刻意保持簡單，而提交層則維持為安全網。

設計背景：High-volume document generation。

API 面向

`DocumentJobStreamProcessor`

主機端應透過工廠建構，如此授權對品牌的控制永遠不會被留在未接線的狀態：

public static function withBrandingFromLicense(
    RenderEngineInterface $engine,
    OutputCommitterInterface $committer,
    IdempotencyStoreInterface $idempotency,
    CheckpointStoreInterface $checkpoints,
    KeyedStateStoreInterface $state,
    DeadLetterStoreInterface $deadLetters,
    RetryPolicy $retryPolicy,
    ClockInterface $clock,
    EntitlementEvaluator $entitlementEvaluator,
    ?LicenseKey $license,
    ?StreamProcessorProbe $probe = null,
    ?JobCompletionEmitterInterface $emitter = null,
    ?BrandingApplicator $brandingApplicator = null,
): self

$clock 是 Symfony\Component\Clock\ClockInterface（重試的退避會透過它睡眠）。null 授權會 fail-closed 解析為評估版品牌。

單一進入點會處理一次執行並回傳其計數器：

public function process(iterable $manifests, StreamProcessorConfig $config): ProcessingSummary

擲出或失敗於： 當當機安全的前提條件失敗時（一個 crashSafe 執行搭配了非持久的協作元件），或當一個提交含糊時，擲出 NextPDF\Enterprise\Stream\Exception\StreamProcessorException；當 windowSize 超過 engine 的 maxBatchSize() 時擲出 InvalidArgumentException。

`StreamProcessorConfig`

public function __construct(
    public string $runId,
    int $windowSize = 32,
    int $checkpointIntervalJobs = 100,
    public bool $crashSafe = true,
    public bool $emitSkippedCompletions = false,
)

擲出或失敗於： 當 windowSize 或 checkpointIntervalJobs 低於 1 時擲出 InvalidArgumentException。$runId 是穩定、單一寫入者的執行識別碼，用來作為檢查點恢復的鍵。

`ObjectStorageCommitter`

public function __construct(
    private ObjectStorageClientInterface $client,
    private string $scheme,
    private ClockInterface $clock,
) {}

$scheme 指名此 committer 所服務的目標 scheme（例如 s3 或 gcs）；此處的 $clock 是 Psr\Clock\ClockInterface。

public function commit(
    string $jobId,
    OutputObjectKey $target,
    string $bytes,
    string $sha256,
    bool $overwrite = false,
): CommitReceipt

擲出或失敗於： scheme 不符時擲出 UnsupportedTargetException；目標鍵並非 container 相對安全時擲出 RenderManifestException；當宣告的 sha-256 與位元組不符時擲出 CommitIntegrityException；沒有 overwrite 而位元組相異時擲出 OutputCommitConflictException（SPEC-COMMIT-409）；在並行變更下建立競賽於 5 次嘗試後仍無法收斂時擲出 RuntimeException。

`ObjectStorageClientInterface`

實際的 S3/GCS 整合所需實作的最小轉接器面向：

public function shaOf(string $bucket, string $key): ?string;

public function put(string $bucket, string $key, string $bytes, string $sha256): void;

public function putIfAbsent(string $bucket, string $key, string $bytes, string $sha256): bool;

putIfAbsent() 必須是一個真正的原子條件式建立（在 S3 上是 If-None-Match: *，在 GCS 上是 ifGenerationMatch: 0），且僅在這次呼叫確實寫入了該物件時回傳 true。put() 則是無條件的覆寫，僅在 manifest 要求 overwrite 時使用。

`JobCompletionEmitterInterface` 與 `FilesystemOutboxEmitter`

public function emit(JobTerminalEvent $event): void;

emitter 在處理器上是選用的。事件只會在一項項目被持久終結後才觸發。FilesystemOutboxEmitter 是隨附的持久實作：

public function __construct(string $directory, ?AtomicFileWriter $writer = null)

擲出或失敗於： 當目錄不存在時擲出 InvalidArgumentException；若事件無法被 JSON 編碼，emit() 會擲出 RuntimeException。hasEvent(string $eventId): bool 檢查 outbox；count(): int 回報尚未交付的事件。

`JobTerminalEvent` 與 `JobTerminalStatus`

public function __construct(
    public string $eventId,
    public string $runId,
    public int $sourceOffset,
    public string $jobId,
    public string $idempotencyKeyValue,
    public JobTerminalStatus $status,
    public ?CommitReceipt $receipt,
    public ?string $errorCode,
    public ?string $errorMessage,
    public int $attempts,
    public DateTimeImmutable $occurredAt,
) {}

eventId 是確定性的——runId:sourceOffset:idempotencyKey:status——這正是使 outbox 去重成為可能的原因。toArray() 為傳輸序列化此事件；它不攜帶任何 PDF 位元組。JobTerminalStatus 是一個字串 enum：Committed（committed）、DeadLettered（dead_lettered）、Skipped（skipped）。

`ProcessingSummary`

由 process() 回傳的不可變計數器：runId、sourceRead、fastForwardedByCheckpoint、skippedByIdempotency、windows、renderBatchCalls、renderRetries、commitReceipts、deadLettered、checkpointSaves，以及 finalCommittedOffset（最終的終態高水位）。

程式碼範例——快速開始

單獨呈現 exactly-once 的物件儲存提交。記憶體內的 NullObjectStorageClient 代替你的 S3/GCS 轉接器；你所觀察到的語意，正是實際轉接器必須維持的那些。

<?php

declare(strict_types=1);

require __DIR__ . '/vendor/autoload.php';

use NextPDF\Enterprise\Stream\Storage\NullObjectStorageClient;
use NextPDF\Enterprise\Stream\Storage\ObjectStorageCommitter;
use NextPDF\Manifest\OutputObjectKey;
use NextPDF\Pro\Stream\Exception\OutputCommitConflictException;
use Symfony\Component\Clock\NativeClock;

$committer = new ObjectStorageCommitter(
    client: new NullObjectStorageClient(), // swap in your S3/GCS adapter
    scheme: 's3',
    clock: new NativeClock(),
);

$target = new OutputObjectKey(scheme: 's3', container: 'invoices', key: '2026/07/inv-1001.pdf');
$bytes  = '%PDF-1.7 example-rendered-bytes';
$sha    = hash('sha256', $bytes);

$first  = $committer->commit('inv-1001', $target, $bytes, $sha);
$replay = $committer->commit('inv-1001', $target, $bytes, $sha); // crash-resume replay

printf("first : reuse=%s, %d bytes\n", var_export($first->idempotentReuse, true), $first->bytesWritten);
printf("replay: reuse=%s\n", var_export($replay->idempotentReuse, true));

try {
    $divergent = '%PDF-1.7 different-bytes';
    $committer->commit('inv-1001', $target, $divergent, hash('sha256', $divergent));
} catch (OutputCommitConflictException $conflict) {
    echo 'conflict: ' . $conflict->specCode() . "\n"; // no silent clobber
}

預期輸出：

first : reuse=false, 31 bytes
replay: reuse=true
conflict: SPEC-COMMIT-409

程式碼範例——正式環境

一次完整的當機安全執行：持久的 Pro 儲存、物件儲存提交器、一個持久 outbox，以及授權解析的品牌。當機後以相同 runId 重新執行會快轉並收斂。

<?php

declare(strict_types=1);

require __DIR__ . '/vendor/autoload.php';

use NextPDF\Enterprise\Licensing\EntitlementEvaluator;
use NextPDF\Enterprise\Stream\DocumentJobStreamProcessor;
use NextPDF\Enterprise\Stream\Exception\StreamProcessorException;
use NextPDF\Enterprise\Stream\FilesystemOutboxEmitter;
use NextPDF\Enterprise\Stream\Storage\ObjectStorageCommitter;
use NextPDF\Enterprise\Stream\StreamProcessorConfig;
use NextPDF\Manifest\Render\SingleDocumentRenderer;
use NextPDF\Manifest\RenderManifest;
use NextPDF\Pro\Stream\Checkpoint\FilesystemCheckpointStore;
use NextPDF\Pro\Stream\Dedup\FilesystemIdempotencyStore;
use NextPDF\Pro\Stream\Engine\InProcessRenderEngine;
use NextPDF\Pro\Stream\Retry\FilesystemDeadLetterStore;
use NextPDF\Pro\Stream\Retry\RetryPolicy;
use NextPDF\Pro\Stream\State\InMemoryKeyedStateStore;
use Symfony\Component\Clock\NativeClock;

// Production requires a host-supplied adapter whose putIfAbsent() is a TRUE
// atomic conditional create (S3 If-None-Match: *, GCS ifGenerationMatch: 0)
// and whose shaOf() reads durable object state. NullObjectStorageClient is
// for the quick start only - it keeps nothing across processes.
$s3Client     = new \Aws\S3\S3Client(['region' => 'eu-central-1', 'version' => 'latest']);
$objectClient = new \Acme\Storage\S3ObjectStorageClient($s3Client); // implements ObjectStorageClientInterface

$stateDir = '/var/lib/nextpdf/stream';
foreach (['checkpoints', 'idempotency', 'dead-letters', 'outbox'] as $sub) {
    if (!is_dir($stateDir . '/' . $sub)) {
        mkdir($stateDir . '/' . $sub, 0770, true);
    }
}

// One manifest per JSONL line; the generator never materialises the batch.
$manifests = (static function (string $path): Generator {
    $handle = fopen($path, 'rb');
    if ($handle === false) {
        throw new RuntimeException('Cannot open job stream: ' . $path);
    }
    try {
        while (($line = fgets($handle)) !== false) {
            if (trim($line) !== '') {
                yield RenderManifest::fromJson(trim($line));
            }
        }
    } finally {
        fclose($handle);
    }
})('/var/spool/nextpdf/jobs.jsonl');

$license = null; // your licensing bootstrap yields a LicenseKey; null = evaluation branding

$processor = DocumentJobStreamProcessor::withBrandingFromLicense(
    engine: new InProcessRenderEngine(SingleDocumentRenderer::standalone()),
    // For a live bucket, implement ObjectStorageClientInterface over your S3/GCS SDK.
    committer: new ObjectStorageCommitter($objectClient, 's3', new NativeClock()),
    idempotency: new FilesystemIdempotencyStore($stateDir . '/idempotency'),
    checkpoints: new FilesystemCheckpointStore($stateDir . '/checkpoints'),
    state: new InMemoryKeyedStateStore(), // recomputable; durability not required here
    deadLetters: new FilesystemDeadLetterStore($stateDir . '/dead-letters'),
    retryPolicy: new RetryPolicy(maxAttempts: 3, baseDelayMs: 200, maxDelayMs: 5_000),
    clock: new NativeClock(),
    entitlementEvaluator: new EntitlementEvaluator(),
    license: $license,
    emitter: new FilesystemOutboxEmitter($stateDir . '/outbox'),
);

$config = new StreamProcessorConfig(
    runId: 'nightly-invoices-2026-07-03',
    windowSize: 32,
    checkpointIntervalJobs: 100,
    crashSafe: true,
);

try {
    $summary = $processor->process($manifests, $config);
} catch (StreamProcessorException $e) {
    // Ambiguous commit or a non-durable collaborator: the finalized prefix is
    // checkpointed. Re-run the SAME runId; the committer converges by digest.
    fwrite(STDERR, 'Run aborted for safe resume: ' . $e->getMessage() . PHP_EOL);
    exit(1);
}

printf(
    "run %s: read=%d committed=%d dedup-skipped=%d dead-lettered=%d checkpoints=%d final-offset=%d\n",
    $summary->runId,
    $summary->sourceRead,
    $summary->commitReceipts,
    $summary->skippedByIdempotency,
    $summary->deadLettered,
    $summary->checkpointSaves,
    $summary->finalCommittedOffset,
);

範例輸出（計數器取決於你的工作串流）：

run nightly-invoices-2026-07-03: read=1200 committed=1187 dedup-skipped=13 dead-lettered=0 checkpoints=12 final-offset=1200

邊界情況與陷阱

每個 runId 單一寫入者是你的責任。 檢查點儲存沒有租約或 compare-and-swap。同一個 runId 上的兩個並行寫入者在合約之外；請在你的排程器中強制執行互斥。
crashSafe: true 對非持久協作元件會快速失敗。 committer、checkpoint、idempotency 與 dead-letter 儲存都必須實作 DurableCapability 標記，否則 process() 會擲出 StreamProcessorException 並指名違規者。有鍵狀態儲存刻意豁免：遺失的有鍵狀態會從檢查點往後重新計算。
windowSize 必須符合 engine。 大於 maxBatchSize() 的視窗會在任何工作開始前擲出 InvalidArgumentException。
含糊的提交會中止；衝突則不會。 SPEC-COMMIT-409 是一個確定性的終態衝突：該項目送入死信，執行繼續。任何其他提交失敗都是含糊的：已終結的前綴被記入檢查點，執行則擲出以利安全恢復。
渲染失敗絕不會中止執行。 逐項目的 Failed 結果、耗盡的重試預算，或無法加品牌的評估版位元組，都會將該項目送入死信並繼續。
重複的 jobId 值是安全的；重複的工作以 idempotencyKey 為鍵。 結果以唯一的來源偏移量對應到項目，絕不以 jobId。重複的 idempotency 鍵即使在同一屏障區間內、在重新渲染之前，也會被辨識出來。
emitter 的持久性決定了事件語意。 單純的 callback emitter 僅為觀察者，且跨當機為 at-most-once。FilesystemOutboxEmitter 讓 outbox 成為持久且以去重鍵標記；轉送交付於是為 at-least-once，而下游的 exactly-once 則需要消費端以 eventId 去重。它的目錄（如同每個檔案系統儲存一樣）必須事先存在，否則建構子會擲出 InvalidArgumentException。
略過的事件預設關閉。 設定 emitSkippedCompletions: true 以同時為被去重短路的項目發出一個 Skipped 終態事件。

安全性說明

輸出鍵 fail closed。 commit() 會重新斷言目標鍵為 container 相對安全：沒有 .. 穿越、沒有絕對路徑逃逸、沒有 null 位元組、沒有內嵌的 stream-wrapper scheme，也沒有冒號（冒號會封閉 NTFS 替代資料流向量）。不安全的鍵會在任何儲存呼叫之前擲出。
完整性在邊界被重新驗證。 committer 會針對實際位元組重新計算 sha-256，並以 CommitIntegrityException 拒絕不符者，因此被破壞的交接無法悄然落地。
事件不攜帶任何文件內容。 JobTerminalEvent 與 outbox 列僅持有識別碼、摘要、時間戳與錯誤字串。錯誤訊息可能回響 engine 診斷；在將 outbox 檔案送往第三方接收端之前，請清除它們以及任何可識別租戶的 jobId 命名法。
評估版輸出絕不會在未加品牌下發布。 當需要品牌卻無法套用時，該項目會被送入死信而非提交。
跨寫入者的 exactly-once 只與你的轉接器一樣強。 若 putIfAbsent() 不是真正的原子條件式寫入，該保證便退化為單一寫入者語意。物件儲存憑證與 bucket 政策屬主機端的事務；此模組絕不代管它們。

一致性

沒有任何已發布的標準定義此模組的行為。本頁上的 exactly-once、checkpoint 與 outbox 保證是 NextPDF Enterprise API 的工程合約，在此以外部可觀察的行為陳述——它們並非對任何標準的一致性或認證。內部將 SHA-256 用作完整性摘要同樣是管線設施，而非合規宣稱。一如 NextPDF 各處：支援不等於一致性，一致性也不等於認證。NextPDF 未持有任何認證，也不授予任何認證；建立在此模組之上的部署是否符合你的法規或合約義務，是你的評估者的判定。

行為合約

終態事件並非檢查點交易的一部分：發出在 checkpoint.save() 之後執行，因此 outbox 持久地持有每個已發出的事件，但並非跨當機的完整帳簿。已提交的物件仍是真實來源。

位於 finalCommittedOffset 或其之下的每個來源偏移量都恰好到達一種終態結果：Committed、Skipped 或 DeadLettered。
項目依來源偏移量順序終結；屏障順序是提交、儲存檢查點、沖刷 idempotency 標記，然後才發出事件。
以相同 runId 重新執行絕不會重複發布：已記入檢查點的偏移量會快轉，而位元組相同的重新提交是經摘要比對、idempotentReuse: true 的 no-op。
全新物件只會透過原子條件式建立被建立；在被佔用的鍵上、沒有 overwrite 而位元組相異者，是一個確定性的 SPEC-COMMIT-409 死信，絕不是覆寫。
含糊的提交會將已終結的前綴記入檢查點並以 StreamProcessorException 中止；失敗的偏移量不會被推進。
一個 crashSafe 執行會在讀取任何輸入之前，拒絕非持久的 committer、checkpoint、idempotency 或 dead-letter 協作元件。
事件 id 是 run、offset、idempotency 鍵與狀態的純函式，因此持久 outbox 每個事件至多持有一列。

Core 後備方案

NextPDF Core 透過 writer 與 render-manifest 合約一次渲染一份文件——見 Writer。Core 本身沒有持久的工作串流、沒有檢查點恢復、沒有 idempotency 去重、沒有物件儲存提交，也沒有終態事件 outbox。NextPDF Pro 加入持久且可並行的 render engine 以及單一主機的檔案系統儲存（Stream in Pro）。跨主機的那一半——這個處理器、物件儲存提交器與持久 outbox——需要 NextPDF Enterprise。

發布邊界

本頁僅記錄外部可觀察的行為與受支援的公開 API 面向。內部命名空間路徑、輔助類別、機制表格、runbook 檔名與工單前綴皆不在範圍內。

另見

Stream (Pro)——行程內的那一半：render engine、executor 與本機持久儲存。
Stream — Deep Reference——共用 Stream 介面的合約層參考。
Output Pipeline (Enterprise)——在 pipeline manifest 之上的批次編排。
Trial and branding——評估版品牌如何被解析與套用。
High-volume document generation——此模組存在的目的情境。
Operating NextPDF in production——長時間執行 worker 的部署姿態。