Enterprise 版本

Stream：文档作业处理

概览

NextPDF\Enterprise\Stream\DocumentJobStreamProcessor 将一条渲染清单流转化为持久、可追责的结果。它以生成器方式消费 iterable<RenderManifest>，通过 Pro 渲染引擎渲染有界窗口，并按源顺序终结每个作业。每个作业都恰好以一个终态结束：输出已提交、已识别为先前已提交，或被死信。进度会被检查点记录，因此崩溃的运行可以恢复而不会重新发布任何内容。

Stream 的能力分布在两个版本中，且这种划分是刻意为之的。Pro 提供持久、并发的渲染引擎以及本地、单主机的文件系统存储——即进程内的那一半。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 契约——引擎、提交器、检查点、幂等性、重试与死信接口。

概念综述

处理器的职责是投递语义，而非渲染。它把清单流按源偏移分组成不大于引擎批处理大小的窗口。每个窗口通过 RenderEngineInterface::renderBatch() 渲染，并对逐项超时进行有界、确定性的重试。随后每一项都按源偏移顺序终结为一个终态结果。

恰好一次的边界是逐项的，并且锚定在提交器上，而非协调机制上。持久进度是一个从 1 开始的偏移高水位线：每个不超过检查点的偏移都已达到终态结果。屏障顺序是固定的：提交字节、推进水位线、保存检查点、刷新缓冲的幂等标记，然后发出终态事件。提交与检查点之间的崩溃会在恢复时幂等地重新提交，因为提交器会比较摘要。检查点之后的崩溃会快进越过该偏移，因此不会有任何内容被发布两次。

ObjectStorageCommitter 通过最小化的 ObjectStorageClientInterface，针对某个对象存储实现 Pro 的 OutputCommitterInterface。目标的 container 即桶（bucket），其 key 即对象键。重新提交相同的字节是一次经摘要比较的空操作。在没有 overwrite 的情况下提交分歧字节会引发 SPEC-COMMIT-409 冲突。全新对象只会通过原子的 putIfAbsent() 条件式写入来创建；在该竞态中落败会触发一次有界的重读并解决循环。因此，跨写入者的恰好一次成立的程度，恰好取决于你的适配器的 putIfAbsent() 是否为真正的条件式写入——S3 上的 If-None-Match: *、GCS 上的 ifGenerationMatch: 0。本周期发布该接口以及内存版的 NullObjectStorageClient；实时 S3/GCS 适配器由宿主提供。

终态事件为下游系统闭合了回路。在检查点屏障之后，处理器会尝试为每个已终结的作业发出一个 JobTerminalEvent——标识符、状态、回执、错误详情、尝试次数，但绝不包含任何 PDF 字节。使用普通回调发送器时，发出是至多一次的：检查点之后的事件在崩溃恢复时可能被跳过。FilesystemOutboxEmitter 使得一旦 emit() 运行，每个事件都持久化：每个事件是一个以其确定性 eventId 的哈希命名的原子 JSON 文件，因此恢复后的重新发出是幂等的，中继会至少投递一次，消费者则基于 eventId 去重。无论哪种方式,有一条边界始终存在：发出发生在检查点屏障之后，因此 checkpoint.save() 与 emit() 之间的崩溃会在恢复时跳过该项的终态事件。需要完整事件账本的下游系统应当对照已提交的对象进行核对（存储是真相之源），而不是仅依赖 outbox。

授权被接入到输出路径中。withBrandingFromLicense() 工厂在每次运行时从授权解析一次评估版品牌化策略。付费授权解析为恒等变换。评估版或缺失的授权会为每个已提交文档加上水印，而无法被品牌化的文档会被死信——处理器绝不会提交未品牌化的评估版字节。

为何如此设计

关键的承重决策是：恰好一次依托于提交器那经摘要比较、条件式创建的对象——而非分布式锁或共识。对象存储的条件式写入是本设计所需的唯一原子原语，其余一切都被允许失败并恢复。这就是为什么渲染引擎必须保持无副作用，为什么键控状态和运行内去重缓存被视作可重新计算的加速项，以及为什么一次模棱两可的提交会中止运行而非猜测：恢复路径会通过同一次摘要比较收敛。这也是为什么每个 runId 单写入者是一项明示的要求，而非强制的租约——检查点存储刻意保持简单，而提交层保持为安全网。

设计背景：大批量文档生成。

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 授权会失败关闭地解析为评估版品牌化。

单一入口点处理一次运行并返回其计数器：

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

抛出或失败于： 当崩溃安全前提条件不满足（crashSafe 运行搭配非持久协作组件）或提交模棱两可时，抛出 NextPDF\Enterprise\Stream\Exception\StreamProcessorException；当 windowSize 超过引擎的 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 命名此提交器所服务的目标 scheme（例如 s3 或 gcs）；这里的 $clock 为 Psr\Clock\ClockInterface。

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

抛出或失败于： scheme 不匹配时抛出 UnsupportedTargetException；目标键不是容器相对安全时抛出 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() 是无条件覆盖，仅在清单请求了 overwrite 时使用。

`JobCompletionEmitterInterface` 与 `FilesystemOutboxEmitter`

public function emit(JobTerminalEvent $event): void;

发送器在处理器上是可选的。事件仅在一项被持久终结之后才触发。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 是一个字符串枚举：Committed（committed）、DeadLettered（dead_lettered）、Skipped（skipped）。

`ProcessingSummary`

由 process() 返回的不可变计数器：runId、sourceRead、fastForwardedByCheckpoint、skippedByIdempotency、windows、renderBatchCalls、renderRetries、commitReceipts、deadLettered、checkpointSaves，以及 finalCommittedOffset（最终的终态高水位线）。

代码示例——快速上手

孤立地演示恰好一次的对象存储提交。内存版的 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 的单写入者是你的责任。 检查点存储没有租约或比较并交换。在一个 runId 上同时存在两个写入者超出了契约范围；请在你的调度器中强制互斥。
crashSafe: true 在非持久协作组件上快速失败。 提交器、检查点、幂等性与死信存储都必须实现 DurableCapability 标记，否则 process() 抛出 StreamProcessorException 并指名违规者。键控状态存储被刻意豁免：丢失的键控状态会从检查点向前重新计算。
windowSize 必须适配引擎。 大于 maxBatchSize() 的窗口会在任何工作开始之前抛出 InvalidArgumentException。
模棱两可的提交会中止；冲突则不会。 SPEC-COMMIT-409 是一个确定性的终态冲突：该项被死信，运行继续。任何其他提交失败都是模棱两可的：已终结的前缀被检查点记录，运行则抛出以便安全恢复。
渲染失败绝不会中止运行。 逐项的 Failed 结果、耗尽的重试预算，或无法品牌化的评估版字节，都会将该项死信并继续。
重复的 jobId 值是安全的；重复的工作以 idempotencyKey 为键。 结果通过唯一的源偏移与各项关联，绝不通过 jobId。重复的幂等键即使在同一屏障间隔内、在重新渲染之前也会被识别。
发送器的持久性决定事件语义。 普通回调发送器仅为观察者且跨崩溃为至多一次。FilesystemOutboxEmitter 使 outbox 持久化并以去重键索引；届时中继投递为至少一次，而下游的恰好一次需要消费者基于 eventId 去重。它的目录（与每一个文件系统存储一样）必须预先存在，否则构造函数抛出 InvalidArgumentException。
Skipped 事件默认关闭。 设置 emitSkippedCompletions: true 以额外为经去重短路的项发出 Skipped 终态事件。

安全须知

输出键失败关闭。 commit() 会重新断言目标键是容器相对安全的：没有 .. 遍历、没有绝对路径逃逸、没有空字节、没有嵌入的流包装器 scheme，也没有冒号（冒号会关闭 NTFS 备用数据流向量）。不安全的键会在任何存储调用之前抛出。
完整性在边界处重新验证。 提交器对实际字节重新计算 sha-256 并以 CommitIntegrityException 拒绝不匹配，因此损坏的交接无法悄无声息地落地。
事件不携带文档内容。 JobTerminalEvent 和 outbox 行只保存标识符、摘要、时间戳与错误字符串。错误消息可能回显引擎诊断信息；在将 outbox 文件发送到第三方接收端之前，请清洗它们，以及任何可标识租户的 jobId 命名方案。
评估版输出绝不会未品牌化地发布。 当需要品牌化却无法应用时，该项会被死信而非提交。
跨写入者的恰好一次仅与你的适配器一样强。 如果 putIfAbsent() 不是真正的原子条件式写入，该保证会退化为单写入者语义。对象存储凭据与桶策略属于宿主的职责；本模块从不管理它们。

合规性

没有任何已发布的标准定义本模块的行为。本页所述的恰好一次、检查点与 outbox 保证是 NextPDF Enterprise API 的工程契约，在此作为外部可观察的行为陈述——它们不是对任何标准的合规，也不是针对任何标准的认证。内部将 SHA-256 用作完整性摘要同样是底层管道，而非合规性主张。正如 NextPDF 处处所述：支持不是合规，合规不是认证。NextPDF 不持有任何认证，也不授予任何认证；基于本模块构建的部署是否满足你的监管或合同义务，是由你的评估方作出的判定。

行为契约

终态事件不属于检查点事务：发出运行在 checkpoint.save() 之后，因此 outbox 会持久保存每一个已发出的事件，但跨崩溃并非完整的账本。已提交的对象仍是真相之源。

每个不超过 finalCommittedOffset 的源偏移都恰好达到一个终态结果：Committed、Skipped 或 DeadLettered。
各项按源偏移顺序终结；屏障顺序为提交、检查点保存、幂等标记刷新，然后事件发出。
以相同 runId 重新运行绝不会重复发布：已检查点的偏移会快进，而字节相同的重新提交是经摘要比较的空操作，idempotentReuse: true。
全新对象只会通过原子条件式创建来创建；在已占用的键上、无 overwrite 的分歧字节是一次确定性的 SPEC-COMMIT-409 死信，绝不是覆盖。
模棱两可的提交会检查点记录已终结的前缀并以 StreamProcessorException 中止；失败的偏移不会被推进。
crashSafe 运行会在读取任何输入之前，拒绝非持久的提交器、检查点、幂等性或死信协作组件。
事件 id 是运行、偏移、幂等键与状态的纯函数，因此持久 outbox 对每个事件至多保存一行。

Core 回退

NextPDF Core 通过 writer 与渲染清单契约一次渲染一个文档——参见 Writer。Core 本身没有持久作业流、没有检查点恢复、没有幂等去重、没有对象存储提交，也没有终态事件 outbox。NextPDF Pro 增加了持久、并发的渲染引擎以及单主机文件系统存储（Pro 中的 Stream）。跨主机的那一半——本处理器、对象存储提交器与持久 outbox——需要 NextPDF Enterprise。

发布边界

本页仅记录外部可观察的行为以及受支持的公共 API 表面。内部命名空间路径、辅助类、机制表、运行手册文件名与工单前缀均不在范围内。

另见

Stream（Pro）——进程内的那一半：渲染引擎、执行器与本地持久存储。
Stream——深度参考——共享 Stream 接口的契约级参考。
Output Pipeline（Enterprise）——基于管道清单的批处理编排。
试用与品牌化——评估版品牌化如何被解析与应用。
大批量文档生成——本模块存在的场景。
在生产中运维 NextPDF——长时运行 worker 的部署姿态。