Hackathon/GrandPlan
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Phat Files and FlowFile Packaging & Signing design specs

Browse Source 
Initial commit of NiFi prototyping design docs: the existing Phat Files
large-file transfer spec, the new packaging/signing/chunking design, and
the whiteboard sketch (grand_plan.jpg) it was reconciled against.
This commit is contained in:
wall
2026-07-28 14:17:11 +10:00
commit 4008f61ce4
3 changed files with 779 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
2026-07-23-phat-files-design.md
Executable
+328
View File
@@ -0,0 +1,328 @@
# 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/<key>/<etag>` (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/<key>/<etag>/<block-index>` via `PutObject`, and flag that
`<key>/<etag>` 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 `<key>/<etag>`
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
`<key>/<etag>`, 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 `<key>` 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 `<key>` for a given `<etag>`, it also writes a small,
separate marker object at `_phat/completed/<key>/<etag>` (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
`<key>` 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 `<key>/<etag>` (see
[Resend vs. duplicate](#resend-vs-duplicate)). For the block's
`<key>/<etag>`, lists the staging prefix and compares the object count
against `phat.block.total`. If complete:
1. Assemble (or re-assemble) the final object at `<key>` via server-side
`UploadPartCopy` from the staged blocks (no bytes move back through
NiFi) — this overwrites any previously-assembled object for the same
`<key>/<etag>`, 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/<key>/<etag>` — see [Completion markers](#completion-markers).
4. Delete the staging prefix for that `<key>/<etag>`.
### High side: `MonitorS3StagingStalls` (custom processor)
Timer-driven. For each incomplete `<key>/<etag>` 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 `<key>/<etag>`
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/<key>/<etag>` 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/<key>/<etag>/<block-index>` |
| Assemble | High-side S3 | Server-side `UploadPartCopy`, no re-transfer |
| Completion record | High-side S3 | Marker object at `_phat/completed/<key>/<etag>`, 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/<key>/<etag>` 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.