Enterprise edition

Stream: document-job processing

At a glance

NextPDF\Enterprise\Stream\DocumentJobStreamProcessor turns a stream of render manifests into durable, accountable outcomes. It consumes an iterable<RenderManifest> as a generator, renders bounded windows through the Pro render engine, and finalizes every job in source order. Each job ends in exactly one terminal state: output committed, recognized as already committed, or dead-lettered. Progress is checkpointed, so a crashed run resumes without re-publishing anything.

The Stream story splits across two editions, and the split is deliberate. Pro provides the durable, concurrent render engine and the local, single-host filesystem stores — the in-process half. Enterprise provides this document-job stream processor plus the pieces that cross host boundaries: the object-storage committer (ObjectStorageCommitter) and the durable outbox of terminal events (FilesystemOutboxEmitter). The Pro Stream page states the same boundary from its side.

Availability & licensing

This capability ships in NextPDF Enterprise (nextpdf/enterprise) and activates with an Enterprise-tier license envelope. A deployment without that entitlement does not load the capability’s classes. Compare editions and get a license.

Install

composer require nextpdf/enterprise

The classes on this page live under NextPDF\Enterprise\Stream and NextPDF\Enterprise\Stream\Storage. They consume the frozen Pro contracts in NextPDF\Pro\Stream — engine, committer, checkpoint, idempotency, retry, and dead-letter interfaces.

Conceptual overview

The processor’s job is delivery semantics, not rendering. It groups the manifest stream into source-offset windows no larger than the engine’s batch size. Each window renders through RenderEngineInterface::renderBatch(), with bounded, deterministic retry of per-item timeouts. Then every item is finalized in source-offset order to a terminal outcome.

The exactly-once boundary is per item, and it is anchored in the committer, not in coordination. Durable progress is a 1-based offset high-watermark: every offset at or below the checkpoint has reached a terminal outcome. The barrier order is fixed: commit the bytes, advance the watermark, save the checkpoint, flush buffered idempotency marks, then emit terminal events. A crash between commit and checkpoint re-commits idempotently on resume, because the committer compares digests. A crash after the checkpoint fast-forwards past the offset, so nothing publishes twice.

ObjectStorageCommitter implements the Pro OutputCommitterInterface against an object store through the minimal ObjectStorageClientInterface. The target’s container is the bucket and its key the object key. Re-committing identical bytes is a digest-compared no-op. Divergent bytes without overwrite raise the SPEC-COMMIT-409 conflict. A fresh object is only ever created with the atomic putIfAbsent() conditional write; losing that race triggers a bounded re-read-and-resolve loop. Cross-writer exactly-once therefore holds exactly as far as your adapter’s putIfAbsent() is a true conditional write — If-None-Match: * on S3, ifGenerationMatch: 0 on GCS. This cycle ships the interface plus the in-memory NullObjectStorageClient; the live S3/GCS adapter is host-supplied.

Terminal events close the loop for downstream systems. After the checkpoint barrier the processor attempts to emit a JobTerminalEvent for each finalized job — identifiers, status, receipt, error details, attempt count, and never any PDF bytes. With a plain callback emitter, emission is at-most-once: events after a checkpoint can be skipped on crash-resume. FilesystemOutboxEmitter makes every event durable once emit() runs: each event is one atomic JSON file named by a hash of its deterministic eventId, so re-emitting after a resume is idempotent, a relay delivers at-least-once, and consumers dedup on eventId. One boundary remains either way: emission happens after the checkpoint barrier, so a crash between checkpoint.save() and emit() skips that item’s terminal event on resume. Downstream systems that require a complete event ledger should reconcile against the committed objects (the store is the source of truth), not against the outbox alone.

Licensing is wired into the output path. The withBrandingFromLicense() factory resolves an evaluation-branding strategy once per run from the license. A paid license resolves to an identity transform. An evaluation or missing license watermarks every committed document, and a document that cannot be branded is dead-lettered — the processor never commits unbranded evaluation bytes.

Why it works this way

The load-bearing decision is that exactly-once rests on the committer’s digest-compared, conditionally created object — not on distributed locks or consensus. The object store’s conditional write is the only atomic primitive the design requires, and everything else is allowed to fail and recover. That is why the render engine must stay side-effect-free, why keyed state and in-run dedup caches are treated as recomputable accelerations, and why an ambiguous commit aborts the run instead of guessing: the resume path converges through the same digest comparison. It is also why single-writer per runId is a stated requirement rather than an enforced lease — the checkpoint store deliberately stays simple, and the commit layer stays the safety net.

Design background: High-volume document generation.

API surface

`DocumentJobStreamProcessor`

Hosts should construct through the factory, so the license-to-branding control is never left unwired:

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 is Symfony\Component\Clock\ClockInterface (the retry backoff sleeps through it). A null license resolves fail-closed to evaluation branding.

The single entry point processes one run and returns its counters:

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

Throws or fails with: NextPDF\Enterprise\Stream\Exception\StreamProcessorException when crash-safety preconditions fail (a crashSafe run with non-durable collaborators) or when a commit is ambiguous; InvalidArgumentException when windowSize exceeds the engine’s maxBatchSize().

`StreamProcessorConfig`

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

Throws or fails with: InvalidArgumentException when windowSize or checkpointIntervalJobs is below 1. $runId is the stable, single-writer run identifier that keys checkpoint resume.

`ObjectStorageCommitter`

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

$scheme names the target scheme this committer services (for example s3 or gcs); $clock here is Psr\Clock\ClockInterface.

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

Throws or fails with: UnsupportedTargetException on a scheme mismatch; RenderManifestException when the target key is not container-relative-safe; CommitIntegrityException when the declared sha-256 does not match the bytes; OutputCommitConflictException (SPEC-COMMIT-409) on divergent bytes without overwrite; RuntimeException when the create race cannot converge after 5 attempts under concurrent mutation.

`ObjectStorageClientInterface`

The minimal adapter surface a live S3/GCS integration implements:

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() must be a real atomic conditional create (If-None-Match: * on S3, ifGenerationMatch: 0 on GCS) and returns true only when this call wrote the object. put() is the unconditional overwrite used solely when the manifest requested overwrite.

`JobCompletionEmitterInterface` and `FilesystemOutboxEmitter`

public function emit(JobTerminalEvent $event): void;

The emitter is optional on the processor. Events fire only after an item is durably finalized. FilesystemOutboxEmitter is the shipped durable implementation:

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

Throws or fails with: InvalidArgumentException when the directory does not exist; emit() throws RuntimeException if an event cannot be JSON-encoded. hasEvent(string $eventId): bool checks the outbox; count(): int reports undelivered events.

`JobTerminalEvent` and `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 is deterministic — runId:sourceOffset:idempotencyKey:status — which is what makes outbox dedup possible. toArray() serializes the event for transport; it carries no PDF bytes. JobTerminalStatus is a string enum: Committed (committed), DeadLettered (dead_lettered), Skipped (skipped).

`ProcessingSummary`

Immutable counters returned by process(): runId, sourceRead, fastForwardedByCheckpoint, skippedByIdempotency, windows, renderBatchCalls, renderRetries, commitReceipts, deadLettered, checkpointSaves, and finalCommittedOffset (the final terminal high-watermark).

Code sample — Quick start

Exactly-once object-storage commit in isolation. The in-memory NullObjectStorageClient stands in for your S3/GCS adapter; the semantics you observe are the ones a live adapter must preserve.

<?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
}

Expected output:

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

Code sample — Production

A full crash-safe run: durable Pro stores, the object-storage committer, a durable outbox, and license-resolved branding. Re-running the same runId after a crash fast-forwards and converges.

<?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,
);

Example output (counters depend on your job stream):

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

Edge cases & gotchas

Single-writer per runId is your responsibility. The checkpoint store has no lease or compare-and-swap. Two concurrent writers on one runId are outside the contract; enforce exclusivity in your scheduler.
crashSafe: true fails fast on non-durable collaborators. The committer, checkpoint, idempotency, and dead-letter stores must all implement the DurableCapability marker, or process() throws StreamProcessorException naming the offenders. The keyed state store is deliberately exempt: lost keyed state is recomputed from the checkpoint forward.
windowSize must fit the engine. A window larger than maxBatchSize() throws InvalidArgumentException before any work starts.
An ambiguous commit aborts; a conflict does not. SPEC-COMMIT-409 is a deterministic terminal conflict: the item dead-letters and the run continues. Any other commit failure is ambiguous: the finalized prefix is checkpointed and the run throws for safe resume.
Render failures never abort the run. A per-item Failed result, an exhausted retry budget, or unbrandable evaluation bytes all dead-letter that item and continue.
Duplicate jobId values are safe; duplicate work is keyed off idempotencyKey. Results correlate to items by unique source offset, never by jobId. A duplicate idempotency key is recognized even within the same barrier interval, before re-rendering.
Emitter durability decides event semantics. A plain callback emitter is observer-only and at-most-once across a crash. FilesystemOutboxEmitter makes the outbox durable and dedup-keyed; relay delivery is then at-least-once, and downstream exactly-once requires consumer dedup on eventId. Its directory (like every filesystem store’s) must pre-exist, or the constructor throws InvalidArgumentException.
Skipped events are off by default. Set emitSkippedCompletions: true to also emit a Skipped terminal event for dedup-short-circuited items.

Security notes

Output keys fail closed. commit() re-asserts the target key is container-relative-safe: no .. traversal, no absolute escape, no null byte, no embedded stream-wrapper scheme, and no colon (which closes the NTFS alternate-data-stream vector). Unsafe keys throw before any storage call.
Integrity is re-verified at the boundary. The committer recomputes sha-256 over the actual bytes and rejects a mismatch with CommitIntegrityException, so a corrupted handoff cannot land silently.
Events carry no document content. JobTerminalEvent and the outbox rows hold identifiers, digests, timestamps, and error strings only. Error messages can echo engine diagnostics; scrub them, and any tenant-identifying jobId scheme, before shipping outbox files to third-party sinks.
Evaluation output is never published unbranded. When branding is required and cannot be applied, the item is dead-lettered rather than committed.
Cross-writer exactly-once is only as strong as your adapter. If putIfAbsent() is not a true atomic conditional write, the guarantee degrades to single-writer semantics. Object-store credentials and bucket policy are host concerns; the module never manages them.

Conformance

The exactly-once, checkpoint, and outbox guarantees on this page are engineering contracts of the NextPDF Enterprise API, stated here as externally observable behavior. The internal use of SHA-256 as an integrity digest is likewise plumbing.

Behavior contract

Terminal events are not part of the checkpoint transaction: emission runs after checkpoint.save(), so the outbox holds every emitted event durably but is not a complete ledger across crashes. Committed objects remain the source of truth.

Every source offset at or below finalCommittedOffset has reached exactly one terminal outcome: Committed, Skipped, or DeadLettered.
Items are finalized in source-offset order; the barrier order is commit, checkpoint save, idempotency-mark flush, then event emission.
Re-running a run with the same runId never double-publishes: checkpointed offsets fast-forward, and byte-identical re-commits are digest-compared no-ops with idempotentReuse: true.
A fresh object is only ever created through the atomic conditional create; divergent bytes at an occupied key without overwrite are a deterministic SPEC-COMMIT-409 dead-letter, never a clobber.
An ambiguous commit checkpoints the finalized prefix and aborts with StreamProcessorException; the failed offset is not advanced.
A crashSafe run refuses non-durable committer, checkpoint, idempotency, or dead-letter collaborators before reading any input.
Event ids are a pure function of run, offset, idempotency key, and status, so a durable outbox holds at most one row per event.

Core fallback

NextPDF Core renders one document at a time through the writer and the render-manifest contract — see Writer. Core alone has no durable job streams, no checkpoint resume, no idempotency dedup, no object-storage commit, and no terminal-event outbox. NextPDF Pro adds the durable, concurrent render engine and the single-host filesystem stores (Stream in Pro). The cross-host half — this processor, the object-storage committer, and the durable outbox — requires NextPDF Enterprise.

Publication boundary

This page documents externally observable behavior and the supported public API surface only. Internal namespace paths, helper classes, mechanism tables, runbook filenames, and ticket prefixes are out of scope.