# Phat Files — Design Spec **Date:** 2026-07-23 **Status:** Approved for implementation planning ## Purpose Move very large (TB-scale) files from a low-side S3-compatible bucket to a high-side S3-compatible bucket across a one-way boundary, with no automated return channel. Files are split into fixed-size blocks so that transfers can be resumed by resending only specific missing blocks, without ever needing to restart or re-transmit a whole file. The transport that actually carries a block from low side to high side is intentionally out of scope for this spec — it could be a network relay (NiFi Site-to-Site, HTTP push), physical removable media, or any other one-way carriage mechanism. The design only assumes it's one-way. Packaging a FlowFile (content + attributes) into something portable enough to survive that transport, and reconstructing it on the other side, is handled by NiFi's standard pack/unpack FlowFile processors (e.g. `MergeContent` using the FlowFile Stream packaging format, paired with `UnpackContent` on the receiving side) — not by anything custom in this project. `FetchS3ObjectBlock` and `PutS3ObjectBlock` only ever deal with ordinary FlowFiles; whatever sits between them for the boundary crossing is someone else's stock processor. This is a cross-domain-style transfer: data flows low → high only. Any recovery signal (which blocks are missing) must be produced on the high side and manually carried back down to the low side by an operator — there is no network path from high side back to low side. ## Non-goals - Content-defined/delta chunking for re-syncing *changed* files. Only fixed-size blocks and resend-of-missing-blocks are in scope. - Automated high-to-low signaling of any kind (by design constraint, not an oversight). - Bidirectional transfer (high → low) — this spec covers one direction only. A mirror-image flow could reuse the same processors in principle, but is out of scope here. ## Architecture ``` [Low-side S3] <--Range GET--> [Low-side NiFi] --one-way transport--> [High-side NiFi] --PutObject/UploadPartCopy--> [High-side S3] ^ (mechanism deferred: | | network relay, media, etc.) | resend blocks via missing-block report processor properties (JSON, written by (static, EL/attribute, MonitorS3StagingStalls) or file-driven) | | | +----------------- operator manually carries report file down -------------------------+ ``` - **Low-side NiFi**: reads byte ranges of a file from low-side S3 via ranged `GetObject` (streaming read, never buffers a whole file), splits into fixed-size blocks, and hands each block FlowFile to a stock pack/unpack pair (e.g. `MergeContent` + `UnpackContent`) that carries it one-way across whatever transport is chosen for the boundary and reconstructs it as a FlowFile again on the high side. - **High-side NiFi**: receives blocks, stages each as its own object in high-side S3 (streaming `PutObject`), and once a file's full block set is present, assembles the real object server-side via `UploadPartCopy` (no bytes re-transferred through NiFi) and cleans up staging objects. - **No return channel.** Recovery is manual: the high side detects missing or stalled blocks and writes a JSON report; an operator carries that report down to the low side (per their existing cross-domain process) to drive a targeted resend. ## Components ### Low side: `FetchS3ObjectBlock` (custom processor) Reads a byte range of a source S3 object and emits it as a block FlowFile. **Properties** (all Expression-Language-enabled against FlowFile attributes, so each can be set statically, via `${attribute}`, or driven by an upstream flow that parses a carried-over report file): | Property | Description | |---|---| | Bucket | Source S3 bucket | | Object Key | Source S3 object key | | Block Size | Block size in bytes (configurable; determines byte-range math) | | Block Numbers | `all`, or an explicit list/ranges, e.g. `0-4,7,10-12` | | Resend | Boolean, default `false`. Marks the emitted blocks as a deliberate resend rather than an initial/routine transfer — see [Resend vs. duplicate](#resend-vs-duplicate) below. | **Behavior on trigger:** 1. `HeadObject` to get file size and source ETag. 2. Compute total block count = `ceil(fileSize / blockSize)`. 3. For each requested block index, compute byte range `[index * blockSize, min((index + 1) * blockSize, fileSize))`. 4. Issue a ranged `GetObject` per requested block (streaming, not buffered). 5. Emit one FlowFile per block with attributes: - `phat.object.key` - `phat.source.etag` - `phat.block.index` - `phat.block.total` - `phat.block.size` - `phat.block.checksum` (SHA-256 of the block's content) - `phat.resend` (`true`/`false`, copied from the Resend property) That's it — `FetchS3ObjectBlock` only produces ordinary FlowFiles (content + attributes). Getting them across the boundary intact is the pack/unpack pair's job, not this processor's (see Architecture above). ### Low side: report-driven resend (stock processors) A small flow segment, built from stock NiFi processors, watches a drop location for a carried-over missing-block report file: 1. `GetFile`/`FetchFile` picks up the report JSON at the watched path. 2. `EvaluateJsonPath` extracts `objectKey`, `bucket`, `sourceEtag`, `blockSize`, and `missingBlocks` into FlowFile attributes, and sets `resend = true` unconditionally — a report-driven trigger is by definition always a resend, never an initial transfer. 3. Those attributes feed `FetchS3ObjectBlock`'s EL-bound properties (including its `Resend` property), so the correct blocks are resent automatically — no manual property editing required for the common case. Manual override (typing values directly into `FetchS3ObjectBlock`'s properties, including setting `Resend` to `true`) remains available for ad hoc use. ### High side: `PutS3ObjectBlock` (custom processor) Receives a block FlowFile and stages it. **Behavior:** 1. Verify `phat.block.checksum` against the received content. On mismatch, route to `failure` — never stage corrupted data. 2. If `phat.resend` is **not** `true`: check whether a completion marker exists at `_phat/completed//` (see [Completion markers](#completion-markers) below). If it does, this block is an unrequested duplicate of an already-completed transfer of this exact file version — discard immediately, do not stage. (See [Resend vs. duplicate](#resend-vs-duplicate) below for why this check is skipped when `phat.resend` is `true`.) 3. Otherwise, stream the block to `staging///` via `PutObject`, and flag that `/` for re-assembly regardless of whether it was previously marked complete. Staging by block index makes out-of-order arrival a non-issue (each block writes to its own path regardless of arrival order) and makes duplicates within the same in-progress transfer idempotent (re-writing the same path is harmless). #### Resend vs. duplicate Two situations look identical at the byte level — a block for a `/` whose transfer has already been marked complete (see [Completion markers](#completion-markers) below) — but need opposite handling: - **Unrequested duplicate**: the transport (or an operator) accidentally re-sent a full, unflagged transfer after it already completed — e.g. a retried batch job, or a stray retransmission at the transport layer. This carries no new information and should be silently absorbed. - **Requested resend**: an operator deliberately re-sent specific blocks — either because a missing-block report said so, or because they have independent reason to distrust the already-assembled object (e.g. suspected corruption) — and wants it honored even if the destination believes the transfer already completed. The two are distinguished purely by intent, which only a human (or the report-parsing flow acting on the human's behalf) can supply — nothing about the bytes themselves says which one it is. That's what the `Resend` property on `FetchS3ObjectBlock` is for: it's `false` by default (routine/initial transfers), and the report-driven resend flow and manual ad hoc resends both set it `true`. `PutS3ObjectBlock` trusts that flag: `phat.resend=true` always stages the block and re-triggers `AssembleS3ObjectBlocks` for that `/`, even over a previously-completed one; `phat.resend=false` (or absent) is subject to the duplicate-of-completed short-circuit. This means a resend can correct or re-confirm an already-assembled object, while a plain accidental replay of a finished transfer costs nothing. #### Completion markers Duplicate detection cannot rely on the final assembled object itself, because the whole point of landing it at `` is that something downstream collects it — and typically moves or deletes it afterward. If "transfer complete" were only recorded as a tag on that object, the record would vanish the moment it's collected, and a late unrequested duplicate block arriving after that point would find nothing to match against and get staged as if it were a new transfer. So completion is recorded independently: once `AssembleS3ObjectBlocks` successfully assembles `` for a given ``, it also writes a small, separate marker object at `_phat/completed//` (empty or a tiny JSON blob — content doesn't matter, only presence). This marker is never touched by whatever collects the real deliverable, so duplicate detection in `PutS3ObjectBlock` keeps working regardless of what happens to the object at `` afterward. Markers are small (one per completed file *generation*, not per block), but unlike staging blocks they are not deleted as part of normal completion — left alone, they'd accumulate indefinitely. `CleanupCompletionMarkers` (see below) TTL-expires them on a separate, typically much longer, retention window than staging cleanup, since the tradeoff here is audit/duplicate- protection value vs. storage, not "is this transfer abandoned." ### High side: `AssembleS3ObjectBlocks` (custom processor) Triggered after each block lands (or on a schedule), including re-triggers from a resend against an already-completed `/` (see [Resend vs. duplicate](#resend-vs-duplicate)). For the block's `/`, lists the staging prefix and compares the object count against `phat.block.total`. If complete: 1. Assemble (or re-assemble) the final object at `` via server-side `UploadPartCopy` from the staged blocks (no bytes move back through NiFi) — this overwrites any previously-assembled object for the same `/`, which is the intended effect of a resend. 2. Tag the final object with `x-phat-source-etag` = the source ETag it was built from. 3. Write (or refresh the timestamp on) the completion marker at `_phat/completed//` — see [Completion markers](#completion-markers). 4. Delete the staging prefix for that `/`. ### High side: `MonitorS3StagingStalls` (custom processor) Timer-driven. For each incomplete `/` staging prefix, checks the write time of the most recently staged block. If older than the configurable **Stall Timeout**, raises an event (NiFi bulletin/log/FlowFile) and writes a JSON missing-block report: ```json { "objectKey": "...", "bucket": "...", "sourceEtag": "...", "blockSize": 104857600, "missingBlocks": [3, 17, 42] } ``` Missing block indices are computed directly by diffing the expected range `0..total-1` against what's present in the staging prefix — no separate tracking state needed. **Constraint:** Stall Timeout must be configured shorter than the Cleanup Retention Window below, so operators get a chance to notice and request a resend before the data they've already received is reaped. ### High side: `CleanupStagingBlocks` (custom processor) Timer-driven housekeeping, independent of alerting. Reaps any `/` staging prefix whose most recent write is older than the configurable **Cleanup Retention Window**, so abandoned partial transfers don't grow storage unbounded. ### High side: `CleanupCompletionMarkers` (custom processor) Timer-driven housekeeping for [completion markers](#completion-markers). Deletes any `_phat/completed//` marker older than the configurable **Completion Marker Retention** window. This is a separate, independently-configured window from the Cleanup Retention Window above — staging cleanup is about not hoarding abandoned partial transfers (short timescale), while marker retention is about how long duplicate-of-completed protection stays effective after a transfer succeeds (can be a much longer timescale, since markers are cheap). Once a marker expires, a genuinely late unrequested duplicate for that generation would no longer be recognized as one and would be staged — an accepted tradeoff of choosing a finite retention window over "keep forever." ## Data Flow Summary | Stage | Location | Mechanism | |---|---|---| | Read | Low-side S3 | Ranged `GetObject` (streaming) | | Transport | Low → High | One-way, mechanism deferred (network relay, removable media, etc.) | | Stage | High-side S3 | `PutObject` to `staging///` | | Assemble | High-side S3 | Server-side `UploadPartCopy`, no re-transfer | | Completion record | High-side S3 | Marker object at `_phat/completed//`, independent of the deliverable's lifecycle | | Recovery signal | High side → operator → Low side | JSON report, manually carried | ## Error Handling | Condition | Handling | |---|---| | Checksum mismatch on arrival | Route to `failure`, never staged | | Unrequested duplicate, transfer already completed (`phat.resend=false`) | Detected via presence of `_phat/completed//` marker; discarded silently | | Requested resend, transfer already completed (`phat.resend=true`) | Always staged; re-triggers `AssembleS3ObjectBlocks`, overwriting the previously-assembled object and refreshing its marker | | Duplicate block, transfer in progress | Idempotent overwrite of the same staging path | | Completion marker expired (beyond Completion Marker Retention) | Duplicate protection no longer applies for that generation; a late duplicate would be staged as if new (accepted tradeoff) | | Out-of-order block arrival | No-op by construction (staging keyed by block index) | | Transfer stalled (blocks stopped arriving) | `MonitorS3StagingStalls` raises an alert + JSON report with exact missing block indices, before the Cleanup Retention Window would reap it | | Transfer abandoned beyond retention window | `CleanupStagingBlocks` reaps the staging prefix | ## Testing Following this project family's existing convention (see `nifi-custom-processors`'s `processor-design.md`): pure logic is extracted into package-private static methods so it can be unit-tested without a live NiFi/S3 connection: - Byte-range computation from block size / index / file size - Block-number list/range parsing (`all`, `0-4,7,10-12`) - Staging path construction and completion-marker path construction - Missing-block diffing (expected range vs. staged contents) - Resend-vs-duplicate decision logic (`phat.resend` combined with marker presence/absence) End-to-end behavior (fetch → stage → assemble → marker write → cleanup → stall alert → marker expiry) is verified against the lab's MinIO instance, consistent with how prior processors in this environment were tested. Also worth an explicit test: assemble a transfer, delete/move the final object (simulating downstream collection) exactly as a real consumer would, then confirm an unrequested duplicate block is still correctly discarded and a flagged resend still correctly re-assembles. ## Open Items for Implementation Planning - Exact NAR/module layout (likely mirrors `nifi-custom-processors`: separate low-side and high-side processor modules, or one bundle with both, deployed selectively per NiFi instance). - Controller service reuse for AWS credentials (existing `AwsCredentialsProviderService` pattern from `nifi-custom-processors` should carry over directly). - Concrete default values for Block Size, Stall Timeout, Cleanup Retention Window, and Completion Marker Retention. - **Transport mechanism selection.** Deliberately deferred (see Purpose). Whatever is chosen — NiFi Site-to-Site, HTTP push, or physical media — must be genuinely one-way (note that standard S2S involves a bidirectional peer/site-info handshake even when FlowFile data only moves one direction, so it may not qualify as-is if zero bidirectional connectivity is required). Packaging/unpackaging the FlowFile for that transport is handled by stock NiFi processors (e.g. `MergeContent` + `UnpackContent`), not by anything custom in this project.