Add F3 payload layer: receiver-confirmed completion, receiver budgets, awaitable transfer facade

[YuanTingHsieh]

The complete F3/Cell payload layer the Client API execution modes build on, plus the design revision that documents it. Upper layers (executor backends, trainer engine) consume one primitive:

waiter = downloader.get_waiter()
outcome = waiter.wait(timeout=...)   # returns == delivered

(This PR ships the primitive and the truth model behind it; production callers of get_waiter() arrive with the trainer-engine and executor-backend PRs. Nothing user-visible changes yet.)

How a huge-model send flows: before and after

Before this PR — served is assumed delivered. The sender's books say SUCCESS when the last chunk leaves; whether the receiver could actually store the model is unknowable, and the terminal FINISHED status counts failed receivers too:

sequenceDiagram
    participant S as sender cell (holds 70GB model)
    participant R as receiver cell

    Note over S: flare.send() registers refs, then waits on ad-hoc<br/>machinery: progress messages, timeout guesses, ack grace.
    S->>R: small control message (ref ids, no payload)

    loop receiver-driven chunk pulls
        R->>S: request(R1, state)
        S-->>R: reply(chunk, next state)
    end

    R->>S: request(R1, final state)
    S-->>R: reply(EOF)
    Note over S: sender counts this receiver as SUCCESS<br/>the moment the last chunk is served.

    Note over R: receiver stores the model (e.g. disk write).<br/>If this fails, nobody tells the sender.

    Note over S: transaction FINISHED means receiver COUNT reached,<br/>counting failed receivers too.<br/>No queryable verdict exists afterward.
    Note over S,R: consequences: the sender frees the model or exits its<br/>subprocess on a guess. A receiver whose store failed is<br/>invisible: the workflow proceeds without the model (silent truncation).

Loading

With this PR — served ≠ delivered; the receiver's own stored/failed report decides. Every receiver's outcome is knowable, bounded in time, and queryable afterward:

sequenceDiagram
    participant S as sender cell (holds 70GB model)
    participant R as receiver cell

    Note over S: sender registers transaction T0 with refs R1..Rn.<br/>The model stays in sender memory - only refs travel.<br/>waiter = get_transfer_waiter(T0) is the facade this PR adds.
    S->>R: small control message (ref ids, no payload)

    loop receiver-driven chunk pulls (MBs per chunk)
        R->>S: request(R1, state, CONFIRM_CAPABLE)
        S-->>R: reply(chunk, next state)
    end

    R->>S: request(R1, final state, CONFIRM_CAPABLE)
    S-->>R: reply(EOF, CONFIRM_EXPECTED)
    Note over S: sender notes: all bytes served to this receiver,<br/>outcome pending the receiver's confirmation.<br/>Not yet counted as delivered.

    Note over R: receiver stores the assembled model<br/>(download_completed, e.g. write the file to disk).<br/>This step can fail even though every byte arrived.
    R->>S: CONFIRM: stored OK / store failed (fire and forget)

    Note over S: sender records what the receiver reported.<br/>stored OK: counted as delivered.<br/>store failed: counted as failed, despite the served EOF.
    Note over S: when every expected receiver is counted, the monitor closes T0:<br/>callbacks run, 70GB source freed, THEN the verdict is recorded<br/>(kept 30 min) and waiter.wait() returns - fully settled.

    Note over S,R: bounds on the sender's 5s monitor tick. Budgets are opt-in,<br/>set per transaction or via config vars - acquire needs declared receiver ids.<br/>receiver never pulls: failed at the acquire budget.<br/>receiver goes silent or its CONFIRM is lost: failed at the idle budget.<br/>either budget unset: the transaction TTL remains the backstop.<br/>old-version receiver (no CONFIRM_CAPABLE): counted at EOF, exactly today's behavior.

Loading

The trade-off, stated plainly

Holding the source: the sender keeps the model longer only on the happy path (+ receiver store time + one confirm hop) — exactly the window in which the old code had already freed a model whose delivery was unproven. In the degraded cases the hold is shorter than before:

Path	Before	After
Happy path	freed when last chunk served	+ receiver store + confirm hop
Receiver stalls mid-pull	pinned until full tx TTL	freed at the idle budget (sooner)
Receiver never pulls	pinned until TTL	freed at the acquire budget (sooner)
Absolute worst case	TTL	TTL (unchanged backstop)

The budget rows apply when budgets are configured (opt-in, per transaction or via streaming_receiver_acquire_timeout / streaming_receiver_idle_timeout; the acquire budget additionally requires declared receiver identities). Unconfigured deployments keep exactly the before-column TTL behavior.

Network hiccups: chunk/EOF loss is handled as before (retries, tombstone replay). The one new must-arrive message is the CONFIRM; if it is lost, the receiver is counted FAILED despite a good store — a false negative, bounded by the idle budget, healed by an idempotent retry. This is the Two Generals bound: some message is always last and unconfirmed, so a protocol only chooses which way it errs. Before, we erred toward invisible false success (silent truncation); now we err toward visible, bounded, retryable false failure. Corruption-by-assumption is no longer a possible outcome.

Outcome-recording hardening (follow-up to #4853's F3-1): outcome ownership (_outcome_owners: the transaction entitled to record the verdict for its reused-on-retry tx_id) and _tx_table registration are one atomic step, so concurrent same-id constructors cannot separate the live transaction from its ownership; active tx_id reuse retires the prior transaction with its transfer-facing emissions (progress, done/outcome callbacks) suppressed — their reused tx_id names the live retry; tx required in _record_outcome (redundant flag removed); deep-frozen TransferOutcome (note: receiver_statuses is a MappingProxyType — not JSON/pickle-serializable by design; consumers crossing a boundary materialize with dict(...)); exception-guarded downloaded_to_* callbacks.

Receiver-confirmed completion + retry-aware accounting: a confirm-capable receiver's served EOF/ERROR is provisional; the receiver's fire-and-forget confirmation — carrying its stored-OK / store-failed truth (a finalization failure confirms FAILED) — finalizes it. Receiver truth wins (served-EOF + failed store = FAILED — the disk-offload case); retries overwrite provisional records. Every confirmation echoes a per-serve nonce from the terminal reply, binding it to its exact serve: a stale confirmation from a previous life of a reused ref_id is dropped even when the new life has its own pending serve for the same receiver. Confirmations ride with the download's secure setting. All wire keys optional (both version skews degrade to today's producer-served semantics); per-process kill-switch streaming_receiver_confirm_enabled disables the wire behavior on either side without a code revert.

Declared receiver identities + per-(transfer, receiver) budgets + quorum: when receiver_ids are declared, completion, the aggregate outcome, and the quorum are judged against those identities — a status from an unexpected receiver can never complete a transfer a declared receiver did not get. Budgets (opt-in): a never-pulling receiver fails at the acquire budget (requires declared identities); idleness is judged on transaction-level activity, so a receiver that finished one ref and went silent cannot escape a sibling ref's budgets — with a truth-wins re-check on enforcement. Budget failures resolve the outcome on a monitor pass instead of the TTL, which also bounds lost confirmations (fail-closed). min_receivers/quorum_met give fan-out workflows the k-of-N surface (a quorum receiver must be a declared receiver and succeed on every ref); completed stays the strict all-receivers certificate.

TransferWaiter awaitable facade: event-driven and settle-then-resolve — the outcome is recorded (and waiters released) only after the callback chain and source release complete, so acting on wait() returning can never preempt them. Never hangs (unknown/expired/shut-down ids resolve immediately; shutdown releases all waiters). Optional caller-specified linger (FINISHED-gated) preserves the tombstone replay window; acquired_receivers() is the V1 PAYLOAD_ACQUIRED signal; composes with — never replaces — the existing DOWNLOAD_COMPLETE_CB chain.

Design rev 2.2 + vocabulary (folded from the former #4866): forward-path heartbeat exemption narrowed to the payload-materialization phase via new Topic.TASK_PAYLOAD_READY (heartbeats stay authoritative during training; diagrams re-validated); launch_once=False moves to launch-scoped tokens; F3 dependency/vocabulary-mapping updates; status set to Approved. The implementation-plan doc is removed — it was a point-in-time PR decomposition that execution has already deviated from; the design doc is the durable reference.

Tests: 60+ new unit tests across receiver_confirm_test, receiver_budget_test, transfer_waiter_test, and the F3-1 hardening suite — skew matrix, kill-switch both sides, receiver-truth-wins, retry healing, mixed fleets, tombstone interplay, unsolicited-confirm guard, bounded lost-confirm, multi-ref acquisition, quorum intersection, waiter never-hangs invariants. Full streaming+fobs+legacy sweeps green. Reviewed by adversarial multi-agent workflows (24 + 6 confirmed findings fixed) plus a 4-angle cleanup pass (net −48 LOC, hot-path trims).

[Copilot]

Pull request overview

This PR hardens F3 DownloadService terminal outcome recording and lifecycle handling, focusing on concurrency-safe registration/retirement of transactions, making outcome recording unskippable, and ensuring recorded outcomes are immutable when shared with pollers and callbacks.

Changes:

• Register transaction incarnations before the transaction becomes monitor-visible, and retire still-live prior transactions when a tx_id is reused.
• Remove _accept_outcomes and require tx for _record_outcome() so the live-incarnation guard is always enforced.
• Deep-freeze recorded outcomes (tuple refs + mapping-proxy receiver statuses) and guard downloaded_to_one/downloaded_to_all callbacks against exceptions; add targeted unit tests.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
nvflare/fuel/f3/streaming/download_service.py	Adjusts tx creation/retirement sequencing, enforces incarnation-guarded outcome recording, and wraps serving-path callbacks with exception safety.
nvflare/fuel/f3/streaming/transfer_outcome.py	Deep-freezes outcome containers to prevent mutation of recorded terminal truth by consumers.
tests/unit_test/fuel/f3/streaming/transfer_outcome_test.py	Adds tests for deep-freezing, tx_id reuse retirement behavior, and updated _record_outcome(tx=...) requirement.
tests/unit_test/fuel/f3/streaming/download_test_utils.py	Updates isolated test service state to match production removal of _accept_outcomes.
tests/unit_test/fuel/f3/streaming/download_service_test.py	Replaces _accept_outcomes lock-discipline test with ordering/guard tests; adds serving-path callback exception-safety test.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[greptile-apps]

Greptile Summary

This PR delivers the complete F3/Cell payload layer: receiver-confirmed completion (a receiver's fire-and-forget confirmation replaces producer-served semantics as the terminal truth), per-(transfer, receiver) acquire/idle budgets, and the TransferWaiter awaitable facade over the aggregate TransferOutcome. The design revision (2.2) is folded in and the point-in-time implementation plan doc is removed.

• Receiver-confirmed completion: served EOF/ERROR is provisional for confirm-capable peers; the receiver's confirmation (binding a per-serve nonce to prevent stale cross-life certification) finalizes each receiver's status. Legacy peers degrade gracefully; a process-level kill-switch disables the wire behavior without a code revert.
• Budgets + quorum: optional per-transaction acquire/idle budgets fail-close stalled receivers in bounded time (sooner than the TTL); min_receivers/quorum_met expose a k-of-N surface while completed remains the strict all-receivers certificate.
• TransferWaiter: event-driven, settle-then-resolve \u2014 the outcome is recorded (and waiters released) only after the full callback chain and source release complete, so acting on wait() returning can never preempt cleanup.

Confidence Score: 5/5

Safe to merge; no functional correctness issues found in the concurrency logic, locking invariants, or the receiver-confirmation protocol.

The ownership-and-table atomicity fix, settle-then-resolve ordering in transaction_done, per-serve nonce binding, and budget truth-wins re-check are all implemented correctly and well-covered by adversarial unit tests. Lock ordering is consistent throughout. The two observations are diagnostic gaps, not correctness defects.

download_service.py carries all the concurrency-sensitive state; a second reviewer pass on the budget enforcement snapshot-and-re-check and the _outcome_owners/_tx_waiters co-management would be worthwhile before promoting to a large mixed-fleet deployment.

Important Files Changed

Filename	Overview
nvflare/fuel/f3/streaming/download_service.py	Core payload layer: receiver-confirmed completion, per-receiver budgets, TransferWaiter facade, and hardened ownership atomicity. 823+ additions of concurrency-heavy code; lock ordering (_tx_lock → _outcome_lock) is consistent throughout.
nvflare/fuel/f3/streaming/transfer_outcome.py	Adds deep-frozen TransferOutcome (MappingProxyType receiver_statuses, tuple refs), receiver_ids/min_receivers fields, and quorum_met property. Correctly intersects per-ref successes so no receiver subset satisfies multiple refs.
nvflare/fuel/f3/streaming/obj_downloader.py	Adds receiver_ids, min_receivers, acquire/idle timeout forwarding to DownloadService, and the get_waiter() convenience method. Straightforward pass-through changes.
nvflare/fuel/utils/fobs/decomposers/via_downloader.py	Forwards known_receiver_ids to the downloader transaction, guarded against None-placeholder unknowns. Small targeted change.
tests/unit_test/fuel/f3/streaming/receiver_confirm_test.py	455-line test suite covering skew matrix, kill-switch both sides, receiver-truth-wins, retry healing, nonce-binding, tombstone interplay, and unsolicited-confirm guard.
tests/unit_test/fuel/f3/streaming/receiver_budget_test.py	448-line test suite including concurrent same-id creation, shutdown atomicity with CoordLock harness, multi-ref idle escape, and budget-vs-confirm truth-wins.
tests/unit_test/fuel/f3/streaming/transfer_waiter_test.py	232-line test suite verifying never-hangs invariant across unknown/expired/shutdown ids, settle-then-resolve ordering, linger semantics, and acquired_receivers signal.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant S as Producer (DownloadService)
    participant M as Monitor thread
    participant R as Receiver (download_object)

    Note over S: new_transaction() owns outcome slot atomically
    R->>S: "pull request (CONFIRM_CAPABLE=True)"
    S->>R: chunk reply (data)
    R->>S: "pull request (CONFIRM_CAPABLE=True)"
    S->>R: "terminal reply (EOF, CONFIRM_EXPECTED=True, nonce=N1)"
    Note over S: obj_served() status provisional in _pending_confirms

    alt Happy path
        Note over R: download_completed() succeeds
        R-->>S: "fire-and-forget CONFIRM(SUCCESS, nonce=N1)"
        Note over S: obj_confirmed() finalizes receiver_statuses
        M->>S: transaction_done(FINISHED)
        Note over S: callbacks, release, _record_outcome, waiter resolved COMPLETED
    else Lost confirmation with idle budget
        Note over M: idle budget exhausted, _finalize_receiver(FAILED)
        M->>S: transaction_done(FINISHED)
        Note over S: outcome NOT COMPLETED, receiver recorded FAILED
    else No budget, TTL fires
        M->>S: transaction_done(TIMEOUT)
        Note over S: pending receiver absent from receiver_statuses, NOT COMPLETED
    end

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant S as Producer (DownloadService)
    participant M as Monitor thread
    participant R as Receiver (download_object)

    Note over S: new_transaction() owns outcome slot atomically
    R->>S: "pull request (CONFIRM_CAPABLE=True)"
    S->>R: chunk reply (data)
    R->>S: "pull request (CONFIRM_CAPABLE=True)"
    S->>R: "terminal reply (EOF, CONFIRM_EXPECTED=True, nonce=N1)"
    Note over S: obj_served() status provisional in _pending_confirms

    alt Happy path
        Note over R: download_completed() succeeds
        R-->>S: "fire-and-forget CONFIRM(SUCCESS, nonce=N1)"
        Note over S: obj_confirmed() finalizes receiver_statuses
        M->>S: transaction_done(FINISHED)
        Note over S: callbacks, release, _record_outcome, waiter resolved COMPLETED
    else Lost confirmation with idle budget
        Note over M: idle budget exhausted, _finalize_receiver(FAILED)
        M->>S: transaction_done(FINISHED)
        Note over S: outcome NOT COMPLETED, receiver recorded FAILED
    else No budget, TTL fires
        M->>S: transaction_done(TIMEOUT)
        Note over S: pending receiver absent from receiver_statuses, NOT COMPLETED
    end

Loading

_{Reviews (7): Last reviewed commit: "Make settlement exception-proof; close t..." | Re-trigger Greptile}

[codecov-commenter]

Codecov Report

❌ Patch coverage is 95.50898% with 15 lines in your changes missing coverage. Please review.
✅ Project coverage is 60.46%. Comparing base (fb6ef21) to head (38d56d9).
⚠️ Report is 5 commits behind head on main.

Files with missing lines	Patch %	Lines
nvflare/fuel/f3/streaming/download_service.py	95.30%	14 Missing ⚠️
nvflare/fuel/f3/streaming/obj_downloader.py	50.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4865      +/-   ##
==========================================
+ Coverage   60.28%   60.46%   +0.17%     
==========================================
  Files         971      973       +2     
  Lines       92463    92579     +116     
==========================================
+ Hits        55745    55978     +233     
+ Misses      36718    36601     -117     

Flag	Coverage Δ
unit-tests	60.46% <95.50%> (+0.17%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.