Enterprise edição

Stream: processamento de document-jobs

Em resumo

O NextPDF\Enterprise\Stream\DocumentJobStreamProcessor transforma um stream de render manifests em desfechos duráveis e auditáveis. Ele consome um iterable<RenderManifest> como generator, renderiza janelas limitadas através do Pro render engine e finaliza cada job na ordem da fonte. Cada job termina em exatamente um estado terminal: output committed, reconhecido como already committed ou dead-lettered. O progresso é registrado em checkpoint, então uma execução que sofreu crash retoma sem republicar nada.

A história do Stream se divide entre duas edições, e a divisão é deliberada. O Pro fornece o render engine durável e concorrente e os stores locais de sistema de arquivos single-host — a metade in-process. O Enterprise fornece este processador de stream de document-jobs mais as peças que cruzam fronteiras de host: o committer de object-storage (ObjectStorageCommitter) e o outbox durável de eventos terminais (FilesystemOutboxEmitter). A página do Pro Stream declara a mesma fronteira a partir do seu lado.

Disponibilidade e licenciamento

Esta capacidade vem no NextPDF Enterprise (nextpdf/enterprise) e é ativada com um envelope de licença Enterprise-tier. Uma implantação sem essa habilitação não carrega as classes da capacidade. Compare edições e obtenha uma licença.

Instalação

composer require nextpdf/enterprise

As classes desta página vivem sob NextPDF\Enterprise\Stream e NextPDF\Enterprise\Stream\Storage. Elas consomem os contratos Pro congelados em NextPDF\Pro\Stream — interfaces de engine, committer, checkpoint, idempotency, retry e dead-letter.

Visão conceitual

A tarefa do processador é semântica de entrega, não renderização. Ele agrupa o stream de manifestos em janelas por source-offset não maiores que o batch size do engine. Cada janela renderiza através de RenderEngineInterface::renderBatch(), com retry limitado e determinístico de timeouts por item. Em seguida, cada item é finalizado na ordem do source-offset até um desfecho terminal.

A fronteira do exactly-once é por item, e está ancorada no committer, não na coordenação. O progresso durável é um high-watermark de offset baseado em 1: todo offset igual ou abaixo do checkpoint atingiu um desfecho terminal. A ordem da barreira é fixa: fazer commit dos bytes, avançar o watermark, salvar o checkpoint, esvaziar as marcas de idempotency em buffer e então emitir os eventos terminais. Um crash entre o commit e o checkpoint recommita de forma idempotente na retomada, porque o committer compara digests. Um crash após o checkpoint avança rapidamente além do offset, então nada é publicado duas vezes.

O ObjectStorageCommitter implementa a OutputCommitterInterface do Pro contra um object store através da mínima ObjectStorageClientInterface. O container do target é o bucket e sua key é a object key. Recommitar bytes idênticos é um no-op comparado por digest. Bytes divergentes sem overwrite levantam o conflito SPEC-COMMIT-409. Um objeto novo só é criado com o conditional write atômico putIfAbsent(); perder essa corrida dispara um loop limitado de re-read-and-resolve. O exactly-once entre writers, portanto, se sustenta exatamente na medida em que o putIfAbsent() do seu adapter for um conditional write real — If-None-Match: * no S3, ifGenerationMatch: 0 no GCS. Este ciclo entrega a interface mais o NullObjectStorageClient in-memory; o adapter ao vivo de S3/GCS é fornecido pelo host.

Os eventos terminais fecham o ciclo para os sistemas downstream. Após a barreira do checkpoint, o processador tenta emitir um JobTerminalEvent para cada job finalizado — identifiers, status, receipt, detalhes de erro, contagem de tentativas e nunca quaisquer bytes de PDF. Com um emitter de callback simples, a emissão é at-most-once: eventos após um checkpoint podem ser pulados na retomada pós-crash. O FilesystemOutboxEmitter torna cada evento durável assim que emit() roda: cada evento é um arquivo JSON atômico nomeado por um hash do seu eventId determinístico, então reemitir após uma retomada é idempotente, um relay entrega at-least-once e os consumidores fazem dedup pelo eventId. Uma fronteira permanece de qualquer forma: a emissão acontece após a barreira do checkpoint, então um crash entre checkpoint.save() e emit() pula o evento terminal daquele item na retomada. Os sistemas downstream que exigem um ledger de eventos completo devem reconciliar contra os objetos committed (o store é a fonte da verdade), não contra o outbox sozinho.

O licenciamento está conectado ao caminho de output. A factory withBrandingFromLicense() resolve uma estratégia de evaluation-branding uma vez por execução a partir da licença. Uma licença paga resolve para uma transformação de identidade. Uma licença de avaliação ou ausente aplica marca d’água em cada documento committed, e um documento que não pode receber a marca é dead-lettered — o processador nunca faz commit de bytes de avaliação sem marca.

Por que funciona assim

A decisão que sustenta tudo é que o exactly-once repousa sobre o objeto criado condicionalmente e comparado por digest do committer — não sobre distributed locks ou consenso. O conditional write do object store é a única primitiva atômica que o design exige, e tudo o mais tem permissão para falhar e se recuperar. É por isso que o render engine deve permanecer livre de efeitos colaterais, por que o keyed state e os caches de dedup in-run são tratados como acelerações recomputáveis, e por que um commit ambíguo aborta a execução em vez de adivinhar: o caminho de retomada converge através da mesma comparação de digest. É também por isso que single-writer por runId é um requisito declarado em vez de um lease imposto — o checkpoint store deliberadamente permanece simples, e a camada de commit permanece a rede de segurança.

Contexto de design: Geração de documentos em alto volume.

Superfície de API

`DocumentJobStreamProcessor`

Os hosts devem construir através da factory, para que o controle de license-to-branding nunca fique desconectado:

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 (o backoff do retry dorme através dele). Uma licença null resolve fail-closed para evaluation branding.

O único ponto de entrada processa uma execução e retorna seus contadores:

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

Lança ou falha com: NextPDF\Enterprise\Stream\Exception\StreamProcessorException quando pré-condições de crash-safety falham (uma execução crashSafe com colaboradores não duráveis) ou quando um commit é ambíguo; InvalidArgumentException quando windowSize excede o maxBatchSize() do engine.

`StreamProcessorConfig`

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

Lança ou falha com: InvalidArgumentException quando windowSize ou checkpointIntervalJobs está abaixo de 1. $runId é o identificador de execução estável e single-writer que serve de chave para a retomada por checkpoint.

`ObjectStorageCommitter`

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

$scheme nomeia o scheme de target que este committer atende (por exemplo s3 ou gcs); $clock aqui é Psr\Clock\ClockInterface.

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

Lança ou falha com: UnsupportedTargetException em uma incompatibilidade de scheme; RenderManifestException quando a target key não é container-relative-safe; CommitIntegrityException quando o sha-256 declarado não corresponde aos bytes; OutputCommitConflictException (SPEC-COMMIT-409) em bytes divergentes sem overwrite; RuntimeException quando a corrida de create não converge após 5 tentativas sob mutação concorrente.

`ObjectStorageClientInterface`

A superfície mínima de adapter que uma integração ao vivo de S3/GCS implementa:

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() deve ser um conditional create atômico real (If-None-Match: * no S3, ifGenerationMatch: 0 no GCS) e retorna true apenas quando esta chamada escreveu o objeto. put() é o overwrite incondicional usado unicamente quando o manifesto solicitou overwrite.

`JobCompletionEmitterInterface` e `FilesystemOutboxEmitter`

public function emit(JobTerminalEvent $event): void;

O emitter é opcional no processador. Os eventos disparam apenas depois que um item é finalizado de forma durável. FilesystemOutboxEmitter é a implementação durável entregue:

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

Lança ou falha com: InvalidArgumentException quando o diretório não existe; emit() lança RuntimeException se um evento não puder ser codificado em JSON. hasEvent(string $eventId): bool verifica o outbox; count(): int reporta os eventos não entregues.

`JobTerminalEvent` e `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 é determinístico — runId:sourceOffset:idempotencyKey:status — que é o que torna o dedup do outbox possível. toArray() serializa o evento para transporte; ele não carrega bytes de PDF. JobTerminalStatus é um string enum: Committed (committed), DeadLettered (dead_lettered), Skipped (skipped).

`ProcessingSummary`

Contadores imutáveis retornados por process(): runId, sourceRead, fastForwardedByCheckpoint, skippedByIdempotency, windows, renderBatchCalls, renderRetries, commitReceipts, deadLettered, checkpointSaves e finalCommittedOffset (o high-watermark terminal final).

Exemplo de código — Início rápido

Commit de object-storage exactly-once isolado. O NullObjectStorageClient in-memory substitui o seu adapter de S3/GCS; a semântica que você observa é a mesma que um adapter ao vivo precisa preservar.

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

Saída esperada:

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

Exemplo de código — Produção

Uma execução crash-safe completa: stores Pro duráveis, o committer de object-storage, um outbox durável e branding resolvido por licença. Re-executar o mesmo runId após um crash avança rapidamente e converge.

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

Exemplo de saída (os contadores dependem do seu job stream):

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

Casos-limite e armadilhas

Single-writer por runId é sua responsabilidade. O checkpoint store não tem lease nem compare-and-swap. Dois writers concorrentes em um único runId estão fora do contrato; imponha a exclusividade no seu scheduler.
crashSafe: true falha rápido em colaboradores não duráveis. O committer, o checkpoint, o idempotency e o dead-letter stores devem todos implementar o marker DurableCapability, ou process() lança StreamProcessorException nomeando os culpados. O keyed state store é deliberadamente isento: keyed state perdido é recomputado a partir do checkpoint em diante.
windowSize deve caber no engine. Uma janela maior que maxBatchSize() lança InvalidArgumentException antes de qualquer trabalho começar.
Um commit ambíguo aborta; um conflito não. SPEC-COMMIT-409 é um conflito terminal determinístico: o item é dead-lettered e a execução continua. Qualquer outra falha de commit é ambígua: o prefixo finalizado é registrado em checkpoint e a execução lança para uma retomada segura.
Falhas de renderização nunca abortam a execução. Um resultado Failed por item, um orçamento de retry esgotado ou bytes de avaliação não marcáveis, todos fazem dead-letter daquele item e continuam.
Valores de jobId duplicados são seguros; trabalho duplicado tem como chave o idempotencyKey. Os resultados correlacionam-se aos itens por source offset único, nunca por jobId. Uma idempotency key duplicada é reconhecida mesmo dentro do mesmo intervalo de barreira, antes de re-renderizar.
A durabilidade do emitter decide a semântica de eventos. Um emitter de callback simples é apenas observador e at-most-once através de um crash. FilesystemOutboxEmitter torna o outbox durável e com chave de dedup; a entrega por relay é então at-least-once, e o exactly-once downstream exige dedup do consumidor no eventId. Seu diretório (como o de todo filesystem store) deve pré-existir, ou o construtor lança InvalidArgumentException.
Eventos skipped estão desligados por padrão. Defina emitSkippedCompletions: true para também emitir um evento terminal Skipped para itens curto-circuitados por dedup.

Notas de segurança

Output keys falham fechado. commit() reafirma que a target key é container-relative-safe: sem traversal .., sem escape absoluto, sem null byte, sem stream-wrapper scheme embutido e sem dois-pontos (que fecha o vetor de NTFS alternate-data-stream). Chaves inseguras lançam antes de qualquer chamada ao storage.
A integridade é reverificada na fronteira. O committer recomputa o sha-256 sobre os bytes reais e rejeita uma incompatibilidade com CommitIntegrityException, então um handoff corrompido não pode aterrissar silenciosamente.
Os eventos não carregam conteúdo de documento. JobTerminalEvent e as linhas do outbox contêm apenas identifiers, digests, timestamps e strings de erro. As mensagens de erro podem ecoar diagnósticos do engine; higienize-as, e qualquer esquema de jobId que identifique tenant, antes de enviar arquivos de outbox para sinks de terceiros.
Output de avaliação nunca é publicado sem marca. Quando o branding é exigido e não pode ser aplicado, o item é dead-lettered em vez de committed.
O exactly-once entre writers é apenas tão forte quanto o seu adapter. Se putIfAbsent() não for um conditional write atômico real, a garantia se degrada para semântica single-writer. As credenciais do object-store e a política de bucket são preocupações do host; o módulo nunca as gerencia.

Conformidade

Nenhum padrão publicado define o comportamento deste módulo. As garantias de exactly-once, checkpoint e outbox nesta página são contratos de engenharia da API do NextPDF Enterprise, declarados aqui como comportamento observável externamente — não são conformidade a, nem certificação contra, qualquer padrão. O uso interno de SHA-256 como digest de integridade é igualmente encanamento, não uma alegação de compliance. Como em toda parte no NextPDF: suporte não é conformidade, e conformidade não é certificação. O NextPDF não detém certificação e não concede nenhuma; se uma implantação construída sobre este módulo atende às suas obrigações regulatórias ou contratuais é uma determinação dos seus avaliadores.

Contrato de comportamento

Os eventos terminais não fazem parte da transação de checkpoint: a emissão roda após checkpoint.save(), então o outbox contém cada evento emitido de forma durável, mas não é um ledger completo através de crashes. Os objetos committed permanecem a fonte da verdade.

Todo source offset igual ou abaixo de finalCommittedOffset atingiu exatamente um desfecho terminal: Committed, Skipped ou DeadLettered.
Os itens são finalizados na ordem do source-offset; a ordem da barreira é commit, checkpoint save, idempotency-mark flush e então emissão de evento.
Re-executar uma execução com o mesmo runId nunca republica: offsets registrados em checkpoint avançam rapidamente, e recommits byte-idênticos são no-ops comparados por digest com idempotentReuse: true.
Um objeto novo só é criado através do conditional create atômico; bytes divergentes em uma key ocupada sem overwrite são um dead-letter SPEC-COMMIT-409 determinístico, nunca um clobber.
Um commit ambíguo registra em checkpoint o prefixo finalizado e aborta com StreamProcessorException; o offset que falhou não é avançado.
Uma execução crashSafe recusa colaboradores committer, checkpoint, idempotency ou dead-letter não duráveis antes de ler qualquer input.
Os event ids são uma função pura de run, offset, idempotency key e status, então um outbox durável contém no máximo uma linha por evento.

Fallback do Core

O NextPDF Core renderiza um documento por vez através do writer e do contrato de render-manifest — veja Writer. Sozinho, o Core não tem job streams duráveis, nem retomada por checkpoint, nem dedup por idempotency, nem commit de object-storage, nem outbox de eventos terminais. O NextPDF Pro adiciona o render engine durável e concorrente e os stores de sistema de arquivos single-host (Stream no Pro). A metade cross-host — este processador, o committer de object-storage e o outbox durável — exige o NextPDF Enterprise.

Fronteira de publicação

Esta página documenta apenas o comportamento observável externamente e a superfície de API pública suportada. Caminhos de namespace internos, classes auxiliares, tabelas de mecanismo, nomes de arquivo de runbook e prefixos de ticket estão fora de escopo.

Veja também

Stream (Pro) — a metade in-process: render engine, executors e stores duráveis locais.
Stream — Referência aprofundada — a referência em nível de contrato para as interfaces compartilhadas do Stream.
Output Pipeline (Enterprise) — orquestração de batch sobre manifestos de pipeline.
Trial e branding — como o evaluation branding é resolvido e aplicado.
Geração de documentos em alto volume — o cenário para o qual este módulo existe.
Operando o NextPDF em produção — postura de implantação para workers de longa duração.