Enterprise รุ่น

Stream: การประมวลผลงานเอกสาร

ภาพรวมโดยสังเขป

NextPDF\Enterprise\Stream\DocumentJobStreamProcessor เปลี่ยนสตรีมของ render manifests ให้เป็นผลลัพธ์ที่คงทนและตรวจสอบได้ มันบริโภค iterable<RenderManifest> เป็น generator เรนเดอร์หน้าต่างแบบมีขอบเขตผ่านเอนจินเรนเดอร์ Pro และปิดจบทุกงานตามลำดับต้นทาง แต่ละงานจบลงในสถานะปลายทางเพียงหนึ่งเดียวเท่านั้น: เอาต์พุตถูกคอมมิต ถูกรับรู้ว่าคอมมิตไปแล้ว หรือถูก dead-letter ความคืบหน้าถูกทำ checkpoint ไว้ ดังนั้นรันที่ล่มจะเริ่มทำงานต่อได้โดยไม่เผยแพร่สิ่งใดซ้ำ

เรื่องราวของ Stream แยกออกเป็นสองรุ่น และการแยกนี้เป็นเจตนา Pro จัดหาเอนจินเรนเดอร์ที่คงทนและทำงานพร้อมกันได้ พร้อมกับที่เก็บบนระบบไฟล์ท้องถิ่นแบบโฮสต์เดียว — ครึ่งที่อยู่ในโปรเซส Enterprise จัดหาตัวประมวลผลสตรีมงานเอกสารตัวนี้ บวกกับส่วนที่ข้ามขอบเขตโฮสต์: ตัวคอมมิตที่เก็บอ็อบเจกต์ (ObjectStorageCommitter) และ outbox ที่คงทนของเหตุการณ์ปลายทาง (FilesystemOutboxEmitter) หน้า Pro Stream ระบุขอบเขตเดียวกันนี้จากฝั่งของมัน

ความพร้อมใช้งานและการออกใบอนุญาต

ความสามารถนี้มาพร้อมกับ NextPDF Enterprise (nextpdf/enterprise) และเปิดใช้งานด้วย envelope ใบอนุญาตระดับ Enterprise การติดตั้งที่ไม่มีสิทธิ์นั้นจะไม่โหลดคลาสของความสามารถนี้ เปรียบเทียบรุ่นและรับใบอนุญาต

การติดตั้ง

composer require nextpdf/enterprise

คลาสในหน้านี้อยู่ภายใต้ NextPDF\Enterprise\Stream และ NextPDF\Enterprise\Stream\Storage พวกมันบริโภคสัญญา Pro ที่ถูกตรึงไว้ใน NextPDF\Pro\Stream — อินเทอร์เฟซ engine, committer, checkpoint, idempotency, retry และ dead-letter

ภาพรวมเชิงแนวคิด

งานของตัวประมวลผลคือความหมายของการส่งมอบ ไม่ใช่การเรนเดอร์ มันจัดกลุ่มสตรีม manifest ให้เป็นหน้าต่างตาม source-offset ที่ไม่ใหญ่กว่าขนาดแบตช์ของเอนจิน แต่ละหน้าต่างเรนเดอร์ผ่าน RenderEngineInterface::renderBatch() พร้อมการลองใหม่ต่อรายการที่มีขอบเขตและกำหนดผลได้เมื่อเกิด timeout จากนั้นทุกรายการจะถูกปิดจบ ตามลำดับ source-offset ไปสู่ผลลัพธ์ปลายทาง

ขอบเขตครั้งเดียวเป๊ะเป็นแบบต่อรายการ และมันถูกยึดไว้ในตัวคอมมิต ไม่ใช่ในการประสานงาน ความคืบหน้าที่คงทนคือ high-watermark ของ offset ที่เริ่มนับจาก 1: ทุก offset ที่หรือต่ำกว่า checkpoint ได้ไปถึงผลลัพธ์ปลายทางแล้ว ลำดับของกำแพงกั้นถูกกำหนดตายตัว: คอมมิตไบต์ เลื่อน watermark บันทึก checkpoint ล้างเครื่องหมาย idempotency ที่บัฟเฟอร์ไว้ แล้วจึงปล่อยเหตุการณ์ปลายทาง การล่มระหว่างคอมมิตกับ checkpoint จะคอมมิตซ้ำแบบ idempotent เมื่อเริ่มทำงานต่อ เพราะตัวคอมมิตเปรียบเทียบ digest การล่มหลัง checkpoint จะกรอไปข้างหน้าข้าม offset นั้น ดังนั้นจึงไม่มีสิ่งใดเผยแพร่สองครั้ง

ObjectStorageCommitter อิมพลีเมนต์ Pro OutputCommitterInterface กับที่เก็บอ็อบเจกต์ผ่าน ObjectStorageClientInterface ที่เรียบง่ายที่สุด container ของเป้าหมายคือ bucket และ key ของมันคือคีย์อ็อบเจกต์ การคอมมิตไบต์ที่เหมือนกันซ้ำเป็น no-op ที่เปรียบเทียบ digest ไบต์ที่แตกต่างกันโดยไม่มี overwrite จะยกความขัดแย้ง SPEC-COMMIT-409 อ็อบเจกต์ใหม่จะถูกสร้างด้วยการเขียนแบบมีเงื่อนไข putIfAbsent() เชิงอะตอมเท่านั้น การแพ้ในการแข่งขันนั้นจะกระตุ้นลูปอ่านซ้ำและแก้ไขแบบมีขอบเขต ดังนั้นครั้งเดียวเป๊ะข้ามผู้เขียนจึงเป็นจริงเท่าที่ putIfAbsent() ของอะแดปเตอร์ของคุณเป็นการเขียนแบบมีเงื่อนไขที่แท้จริง — If-None-Match: * บน S3, ifGenerationMatch: 0 บน GCS รอบนี้มาพร้อมอินเทอร์เฟซบวกกับ NullObjectStorageClient ในหน่วยความจำ ส่วนอะแดปเตอร์ S3/GCS แบบสดนั้นโฮสต์เป็นผู้จัดหา

เหตุการณ์ปลายทางปิดวงจรให้กับระบบปลายน้ำ หลังกำแพงกั้น checkpoint ตัวประมวลผลพยายามปล่อย JobTerminalEvent สำหรับแต่ละงานที่ปิดจบแล้ว — ตัวระบุ สถานะ ใบรับ รายละเอียดข้อผิดพลาด จำนวนครั้งที่พยายาม และไม่มีไบต์ PDF ใดๆ ด้วยตัวปล่อยแบบ callback ธรรมดา การปล่อยเป็นแบบอย่างมากที่สุดครั้งเดียว: เหตุการณ์หลัง checkpoint อาจถูกข้ามเมื่อเริ่มทำงานต่อจากการล่ม FilesystemOutboxEmitter ทำให้ทุกเหตุการณ์คงทนเมื่อ emit() รัน: แต่ละเหตุการณ์เป็นไฟล์ JSON เชิงอะตอมหนึ่งไฟล์ที่ตั้งชื่อด้วยแฮชของ eventId ที่กำหนดผลได้ ดังนั้นการปล่อยซ้ำหลังการเริ่มทำงานต่อจึงเป็น idempotent รีเลย์ส่งมอบแบบอย่างน้อยหนึ่งครั้ง และผู้บริโภคทำ dedup ด้วย eventId มีขอบเขตหนึ่งที่ยังคงอยู่ไม่ว่าทางใด: การปล่อยเกิดขึ้นหลังกำแพงกั้น checkpoint ดังนั้นการล่มระหว่าง checkpoint.save() กับ emit() จะข้ามเหตุการณ์ปลายทางของรายการนั้นเมื่อเริ่มทำงานต่อ ระบบปลายน้ำที่ต้องการบัญชีเหตุการณ์ที่สมบูรณ์ควรกระทบยอดกับอ็อบเจกต์ที่คอมมิตแล้ว (ที่เก็บคือแหล่งความจริง) ไม่ใช่กับ outbox เพียงอย่างเดียว

การออกใบอนุญาตถูกเดินสายเข้ากับเส้นทางเอาต์พุต แฟกทอรี withBrandingFromLicense() รีโซลฟ์กลยุทธ์การสร้างแบรนด์แบบทดลองใช้หนึ่งครั้งต่อรันจากใบอนุญาต ใบอนุญาตแบบชำระเงินรีโซลฟ์เป็นการแปลงแบบเอกลักษณ์ ใบอนุญาตแบบทดลองใช้หรือที่ขาดหายไปจะใส่ลายน้ำในทุกเอกสารที่คอมมิต และเอกสารที่ไม่สามารถสร้างแบรนด์ได้จะถูก dead-letter — ตัวประมวลผลจะไม่คอมมิตไบต์ทดลองใช้ที่ไม่มีแบรนด์

ทำไมมันจึงทำงานเช่นนี้

การตัดสินใจที่รับน้ำหนักคือครั้งเดียวเป๊ะวางอยู่บนอ็อบเจกต์ที่เปรียบเทียบ digest และสร้างแบบมีเงื่อนไขของตัวคอมมิต — ไม่ใช่บน distributed locks หรือ consensus การเขียนแบบมีเงื่อนไขของที่เก็บอ็อบเจกต์เป็น primitive เชิงอะตอมเพียงอย่างเดียวที่การออกแบบต้องการ และทุกอย่างที่เหลือได้รับอนุญาตให้ล้มเหลวและกู้คืนได้ นั่นคือเหตุผลที่เอนจินเรนเดอร์ต้องปราศจากผลข้างเคียง ทำไมสถานะแบบมีคีย์และแคช dedup ภายในรันจึงถูกปฏิบัติเสมือนเป็นการเร่งความเร็วที่คำนวณใหม่ได้ และทำไมคอมมิตที่กำกวมจึงยกเลิกรันแทนที่จะเดา: เส้นทางการเริ่มทำงานต่อลู่เข้าผ่านการเปรียบเทียบ digest เดียวกัน มันยังเป็นเหตุผลที่ผู้เขียนคนเดียวต่อ runId เป็นข้อกำหนดที่ระบุไว้แทนที่จะเป็น lease ที่บังคับใช้ — ที่เก็บ checkpoint จงใจให้เรียบง่าย และเลเยอร์คอมมิตยังคงเป็นตาข่ายนิรภัย

พื้นหลังการออกแบบ: การสร้างเอกสารปริมาณสูง

พื้นผิว 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

โยนหรือล้มเหลวด้วย: NextPDF\Enterprise\Stream\Exception\StreamProcessorException เมื่อเงื่อนไขล่วงหน้าของความปลอดภัยต่อการล่มล้มเหลว (รัน crashSafe ที่มีตัวร่วมงานไม่คงทน) หรือเมื่อคอมมิตกำกวม; InvalidArgumentException เมื่อ windowSize เกิน maxBatchSize() ของเอนจิน

`StreamProcessorConfig`

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

โยนหรือล้มเหลวด้วย: InvalidArgumentException เมื่อ windowSize หรือ checkpointIntervalJobs ต่ำกว่า 1 $runId คือตัวระบุรันแบบผู้เขียนคนเดียวที่เสถียรซึ่งใช้เป็นคีย์การเริ่มทำงานต่อจาก checkpoint

`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

โยนหรือล้มเหลวด้วย: UnsupportedTargetException เมื่อ scheme ไม่ตรงกัน; RenderManifestException เมื่อคีย์เป้าหมายไม่ปลอดภัยแบบสัมพัทธ์กับ container; CommitIntegrityException เมื่อ sha-256 ที่ประกาศไว้ไม่ตรงกับไบต์; OutputCommitConflictException (SPEC-COMMIT-409) เมื่อไบต์แตกต่างกันโดยไม่มี overwrite; RuntimeException เมื่อการแข่งขันการสร้างไม่สามารถลู่เข้าได้หลังจาก 5 ครั้งภายใต้การกลายพันธุ์พร้อมกัน

`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() ต้องเป็นการสร้างแบบมีเงื่อนไขเชิงอะตอมที่แท้จริง (If-None-Match: * บน S3, ifGenerationMatch: 0 บน GCS) และคืน true ก็ต่อเมื่อการเรียกครั้งนี้เขียนอ็อบเจกต์เท่านั้น put() คือการเขียนทับแบบไม่มีเงื่อนไขที่ใช้เฉพาะเมื่อ manifest ร้องขอ overwrite

`JobCompletionEmitterInterface` และ `FilesystemOutboxEmitter`

public function emit(JobTerminalEvent $event): void;

ตัวปล่อยเป็นทางเลือกบนตัวประมวลผล เหตุการณ์จะยิงหลังจากรายการถูกปิดจบอย่างคงทนแล้วเท่านั้น FilesystemOutboxEmitter คืออิมพลีเมนต์แบบคงทนที่มาพร้อม:

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

โยนหรือล้มเหลวด้วย: InvalidArgumentException เมื่อไดเรกทอรีไม่มีอยู่; emit() โยน RuntimeException หากเหตุการณ์ไม่สามารถเข้ารหัสเป็น JSON ได้ 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 — ซึ่งเป็นสิ่งที่ทำให้ dedup ของ outbox เป็นไปได้ toArray() ทำ serialize เหตุการณ์เพื่อการขนส่ง มันไม่มีไบต์ PDF JobTerminalStatus เป็น string enum: Committed (committed), DeadLettered (dead_lettered), Skipped (skipped)

`ProcessingSummary`

ตัวนับที่เปลี่ยนแปลงไม่ได้ซึ่งคืนโดย process(): runId, sourceRead, fastForwardedByCheckpoint, skippedByIdempotency, windows, renderBatchCalls, renderRetries, commitReceipts, deadLettered, checkpointSaves และ finalCommittedOffset (high-watermark ปลายทางสุดท้าย)

ตัวอย่างโค้ด — เริ่มต้นอย่างรวดเร็ว

การคอมมิตที่เก็บอ็อบเจกต์แบบครั้งเดียวเป๊ะแบบแยกเดี่ยว 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 เป็นความรับผิดชอบของคุณ ที่เก็บ checkpoint ไม่มี lease หรือ compare-and-swap ผู้เขียนสองคนพร้อมกันบน runId เดียวอยู่นอกสัญญา บังคับความเป็นเอกสิทธิ์ใน scheduler ของคุณ
crashSafe: true ล้มเหลวอย่างรวดเร็วเมื่อมีตัวร่วมงานไม่คงทน ตัวคอมมิต checkpoint idempotency และ dead-letter stores ต้องอิมพลีเมนต์เครื่องหมาย DurableCapability ทั้งหมด มิฉะนั้น process() จะโยน StreamProcessorException ที่ระบุชื่อผู้ละเมิด ที่เก็บสถานะแบบมีคีย์ได้รับการยกเว้นโดยเจตนา: สถานะแบบมีคีย์ที่สูญหายจะถูกคำนวณใหม่จาก checkpoint ไปข้างหน้า
windowSize ต้องพอดีกับเอนจิน หน้าต่างที่ใหญ่กว่า maxBatchSize() จะโยน InvalidArgumentException ก่อนงานใดๆ เริ่ม
คอมมิตที่กำกวมยกเลิก; ความขัดแย้งไม่ยกเลิก SPEC-COMMIT-409 เป็นความขัดแย้งปลายทางที่กำหนดผลได้: รายการนั้น dead-letter และรันดำเนินต่อไป ความล้มเหลวของคอมมิตอื่นใดเป็นแบบกำกวม: prefix ที่ปิดจบแล้วถูกทำ checkpoint และรันโยนเพื่อการเริ่มทำงานต่ออย่างปลอดภัย
ความล้มเหลวในการเรนเดอร์ไม่เคยยกเลิกรัน ผลลัพธ์ Failed ต่อรายการ งบประมาณการลองใหม่ที่หมด หรือไบต์ทดลองใช้ที่สร้างแบรนด์ไม่ได้ ล้วน dead-letter รายการนั้นและดำเนินต่อไป
ค่า jobId ซ้ำนั้นปลอดภัย; งานซ้ำถูกทำคีย์ด้วย idempotencyKey ผลลัพธ์สัมพันธ์กับรายการด้วย source offset ที่ไม่ซ้ำ ไม่เคยด้วย jobId คีย์ idempotency ที่ซ้ำจะถูกรับรู้แม้ภายในช่วงกำแพงกั้นเดียวกัน ก่อนการเรนเดอร์ซ้ำ
ความคงทนของตัวปล่อยกำหนดความหมายของเหตุการณ์ ตัวปล่อยแบบ callback ธรรมดาเป็นแบบผู้สังเกตเท่านั้นและอย่างมากที่สุดครั้งเดียวเมื่อข้ามการล่ม FilesystemOutboxEmitter ทำให้ outbox คงทนและมีคีย์ dedup การส่งมอบแบบรีเลย์จึงเป็นแบบอย่างน้อยหนึ่งครั้ง และครั้งเดียวเป๊ะปลายน้ำต้องการ dedup ที่ผู้บริโภคด้วย eventId ไดเรกทอรีของมัน (เช่นเดียวกับที่เก็บบนระบบไฟล์ทุกตัว) ต้องมีอยู่ก่อน มิฉะนั้นคอนสตรัคเตอร์จะโยน InvalidArgumentException
เหตุการณ์ที่ข้ามถูกปิดโดยค่าเริ่มต้น ตั้ง emitSkippedCompletions: true เพื่อปล่อยเหตุการณ์ปลายทาง Skipped สำหรับรายการที่ถูกลัดวงจรด้วย dedup ด้วย

หมายเหตุด้านความปลอดภัย

คีย์เอาต์พุตล้มเหลวแบบปิด commit() ยืนยันซ้ำว่าคีย์เป้าหมายปลอดภัยแบบสัมพัทธ์กับ container: ไม่มีการ traversal ด้วย .. ไม่มีการหลบหนีแบบสัมบูรณ์ ไม่มี null byte ไม่มี scheme ของ stream-wrapper ที่ฝังอยู่ และไม่มีเครื่องหมายโคลอน (ซึ่งปิดช่องทาง NTFS alternate-data-stream) คีย์ที่ไม่ปลอดภัยจะโยนก่อนการเรียกที่เก็บใดๆ
ความสมบูรณ์ถูกตรวจสอบซ้ำที่ขอบเขต ตัวคอมมิตคำนวณ sha-256 ใหม่บนไบต์จริงและปฏิเสธความไม่ตรงกันด้วย CommitIntegrityException ดังนั้นการส่งต่อที่เสียหายจึงไม่สามารถลงจอดได้อย่างเงียบๆ
เหตุการณ์ไม่มีเนื้อหาเอกสาร JobTerminalEvent และแถวของ outbox ถือเพียงตัวระบุ digest ไทม์สแตมป์ และสตริงข้อผิดพลาดเท่านั้น ข้อความข้อผิดพลาดสามารถสะท้อนการวินิจฉัยของเอนจิน จงขัดล้างมัน และ scheme jobId ที่ระบุผู้เช่าใดๆ ก่อนส่งไฟล์ outbox ไปยัง sink ของบุคคลที่สาม
เอาต์พุตทดลองใช้ไม่เคยเผยแพร่โดยไม่มีแบรนด์ เมื่อการสร้างแบรนด์เป็นสิ่งจำเป็นและไม่สามารถทำได้ รายการนั้นจะถูก dead-letter แทนที่จะคอมมิต
ครั้งเดียวเป๊ะข้ามผู้เขียนแข็งแรงเท่ากับอะแดปเตอร์ของคุณเท่านั้น หาก putIfAbsent() ไม่ใช่การเขียนแบบมีเงื่อนไขเชิงอะตอมที่แท้จริง การรับประกันจะเสื่อมลงไปเป็นความหมายแบบผู้เขียนคนเดียว ข้อมูลรับรองของที่เก็บอ็อบเจกต์และนโยบาย bucket เป็นเรื่องของโฮสต์ โมดูลไม่เคยจัดการมัน

ความสอดคล้อง

ไม่มีมาตรฐานที่เผยแพร่ใดกำหนดพฤติกรรมของโมดูลนี้ การรับประกันครั้งเดียวเป๊ะ checkpoint และ outbox ในหน้านี้เป็นสัญญาทางวิศวกรรมของ NextPDF Enterprise API ที่ระบุไว้ที่นี่ในฐานะพฤติกรรมที่สังเกตได้จากภายนอก — มันไม่ใช่ความสอดคล้องกับ หรือการรับรองต่อ มาตรฐานใด การใช้ SHA-256 ภายในเป็น digest ความสมบูรณ์ก็เป็นเพียงระบบท่อ ไม่ใช่การอ้างการปฏิบัติตาม เช่นเดียวกับทุกที่ใน NextPDF: การรองรับไม่ใช่ความสอดคล้อง และความสอดคล้องไม่ใช่การรับรอง NextPDF ไม่ถือการรับรองใดและไม่มอบให้ใด ว่าการติดตั้งที่สร้างบนโมดูลนี้ตรงตามภาระผูกพันด้านกฎระเบียบหรือสัญญาของคุณหรือไม่นั้นเป็นการตัดสินใจของผู้ประเมินของคุณ

สัญญาพฤติกรรม

เหตุการณ์ปลายทางไม่ได้เป็นส่วนหนึ่งของทรานแซกชัน checkpoint: การปล่อยรันหลัง checkpoint.save() ดังนั้น outbox ถือทุกเหตุการณ์ที่ ปล่อยแล้ว อย่างคงทน แต่ไม่ใช่บัญชีที่สมบูรณ์ข้ามการล่ม อ็อบเจกต์ที่คอมมิตแล้วยังคงเป็นแหล่งความจริง

ทุก source offset ที่หรือต่ำกว่า finalCommittedOffset ได้ไปถึงผลลัพธ์ปลายทางเพียงหนึ่งเดียวเท่านั้น: Committed, Skipped หรือ DeadLettered
รายการถูกปิดจบตามลำดับ source-offset; ลำดับของกำแพงกั้นคือ commit, checkpoint save, idempotency-mark flush แล้วจึงปล่อยเหตุการณ์
การรันซ้ำด้วย runId เดิมไม่เคยเผยแพร่สองครั้ง: offset ที่ทำ checkpoint ไว้จะกรอไปข้างหน้า และการคอมมิตซ้ำที่ไบต์เหมือนกันเป็น no-op ที่เปรียบเทียบ digest ด้วย idempotentReuse: true
อ็อบเจกต์ใหม่จะถูกสร้างผ่านการสร้างแบบมีเงื่อนไขเชิงอะตอมเท่านั้น; ไบต์ที่แตกต่างกันที่คีย์ซึ่งถูกครอบครองโดยไม่มี overwrite เป็น SPEC-COMMIT-409 dead-letter ที่กำหนดผลได้ ไม่เคยเป็นการเขียนทับ
คอมมิตที่กำกวมทำ checkpoint prefix ที่ปิดจบแล้วและยกเลิกด้วย StreamProcessorException; offset ที่ล้มเหลวไม่ถูกเลื่อน
รัน crashSafe ปฏิเสธตัวร่วมงาน committer, checkpoint, idempotency หรือ dead-letter ที่ไม่คงทนก่อนอ่านอินพุตใดๆ
event id เป็นฟังก์ชันบริสุทธิ์ของ run, offset, idempotency key และ status ดังนั้น outbox ที่คงทนถือมากที่สุดหนึ่งแถวต่อเหตุการณ์

ทางเลือกสำรอง Core

NextPDF Core เรนเดอร์เอกสารทีละหนึ่งผ่าน writer และสัญญา render-manifest — ดู Writer Core เพียงลำพังไม่มีสตรีมงานที่คงทน ไม่มีการเริ่มทำงานต่อจาก checkpoint ไม่มี dedup แบบ idempotency ไม่มีการคอมมิตที่เก็บอ็อบเจกต์ และไม่มี outbox ของเหตุการณ์ปลายทาง NextPDF Pro เพิ่มเอนจินเรนเดอร์ที่คงทนและทำงานพร้อมกันได้ พร้อมกับที่เก็บบนระบบไฟล์แบบโฮสต์เดียว (Stream ใน Pro) ครึ่งข้ามโฮสต์ — ตัวประมวลผลนี้ ตัวคอมมิตที่เก็บอ็อบเจกต์ และ outbox ที่คงทน — ต้องการ NextPDF Enterprise

ขอบเขตการเผยแพร่

หน้านี้บันทึกเฉพาะพฤติกรรมที่สังเกตได้จากภายนอกและพื้นผิว API สาธารณะที่รองรับเท่านั้น เส้นทาง namespace ภายใน คลาสตัวช่วย ตารางกลไก ชื่อไฟล์ runbook และคำนำหน้าตั๋วอยู่นอกขอบเขต

ดูเพิ่มเติม

Stream (Pro) — ครึ่งที่อยู่ในโปรเซส: เอนจินเรนเดอร์ executors และที่เก็บที่คงทนแบบท้องถิ่น
Stream — Deep Reference — การอ้างอิงระดับสัญญาสำหรับอินเทอร์เฟซ Stream ที่ใช้ร่วมกัน
Output Pipeline (Enterprise) — การประสานงานแบตช์เหนือ pipeline manifests
การทดลองใช้และการสร้างแบรนด์ — วิธีที่การสร้างแบรนด์ทดลองใช้ถูกรีโซลฟ์และนำไปใช้
การสร้างเอกสารปริมาณสูง — สถานการณ์ที่โมดูลนี้มีอยู่เพื่อมัน
การใช้งาน NextPDF ในการผลิตจริง — ท่าทีการติดตั้งสำหรับ worker ที่รันระยะยาว