[client] Report file drop progress, and stop a withdrawn transfer

The receiver reported progress once per file, after the whole body had
been staged: spool.Write drains its reader before returning, so a large
file sat at nothing until it jumped to done. It now stages through a
reader that reports as the bytes land, the same way the sender already
did, and both sides thin their reports to one per percent and one per
200ms so a fast transfer cannot flood the history or the UI.

Progress also becomes an event of its own. It was left out because a
report per byte would have been unusable; thinned, it costs a handful of
events a second and spares every UI a poll. The daemon's event bridge
ignores the new kind, so nothing is published where a notification would
be noise.

Withdrawing consent mid-transfer did nothing. Cancel routed through
Decide, which only moves an offer out of Pending, so an accepted offer
kept its decision, kept its spool, and kept being uploaded into: the
receiver's list went quiet while the sender ran to completion. Revoke
takes an offer back whatever it has already answered, the staging reader
gives up as soon as consent is gone, and the sender stops rather than
retrying a refusal three times over. A transfer withdrawn this way reads
as declined on both ends, which is what it is, rather than as a failure.
This commit is contained in:
Zoltan Papp
2026-08-18 14:34:44 +02:00
parent a94424d563
commit b165e0126e
7 changed files with 262 additions and 20 deletions
+53 -1
View File
@@ -138,7 +138,38 @@ func (r *receiver) upload(sender senderIdentity, id OfferID, index int, offset i
return fmt.Errorf("%w: offset out of range", ErrInvalidOffer)
}
received, err := r.spool.Write(id, index, offset, body, size)
// The decision is read once, at the top, but a whole file goes into this
// one request: a receiver that declines halfway through would otherwise be
// streamed the rest of it, into a spool it has already thrown away.
watched := &acceptedReader{
r: body,
accepted: func() bool {
current, ok := r.offers.Get(sender.key, id)
return ok && current.Decision == DecisionAccepted
},
}
// Write drains the whole body before returning, so without a reader in
// between the only progress the receiver would ever report is the finished
// file. Reports are thinned the same way the sender thins its own.
staged := &progressReader{
r: watched,
sent: offset,
total: size,
report: func(sent int64) {
r.offers.SetProgress(id, index, sent)
r.notifyProgress(offer, index, sent)
},
}
received, err := r.spool.Write(id, index, offset, staged, size)
// A withdrawn offer is an answer, not a failure: the state it moved to is
// the one the user chose, and the spool is already gone.
if errors.Is(err, ErrNotAccepted) {
return err
}
r.offers.SetProgress(id, index, received)
r.notifyProgress(offer, index, received)
@@ -170,6 +201,27 @@ func (r *receiver) close() {
}
}
// acceptedReader stops a staged copy as soon as the offer behind it stops being
// accepted. The check runs on the same cadence as progress rather than on every
// read: it takes the offer store's lock, and a request that keeps going for one
// more chunk after a decline costs nothing.
type acceptedReader struct {
r io.Reader
accepted func() bool
last time.Time
}
func (a *acceptedReader) Read(b []byte) (int, error) {
now := time.Now()
if now.Sub(a.last) >= progressInterval {
a.last = now
if !a.accepted() {
return 0, ErrNotAccepted
}
}
return a.r.Read(b)
}
func (r *receiver) notifyOffer(offer Offer) {
if r.notifier != nil {
r.notifier.OnOffer(offer)