[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.

client/android/filedrop_transfer.go

@@ -46,6 +46,7 @@ const (
 	FileDropEventCompleted = int(filedrop.EventCompleted)
 	FileDropEventFailed    = int(filedrop.EventFailed)
 	FileDropEventWithdrawn = int(filedrop.EventWithdrawn)
+	FileDropEventProgress  = int(filedrop.EventProgress)
 )
 
 // FileDropListener receives transfer events. Calls arrive on background

client/internal/filedrop/client.go

@@ -43,13 +43,6 @@ type ClientConfig struct {
 	OfferTimeout time.Duration
 }
 
-type progressReader struct {
-	r      io.Reader
-	sent   int64
-	total  int64
-	report func(sent int64)
-}
-
 // Client sends offers and payloads to a peer's file drop service.
 type Client struct {
 	http         *http.Client
@@ -58,15 +51,6 @@ type Client struct {
 	offerTimeout time.Duration
 }
 
-func (p *progressReader) Read(b []byte) (int, error) {
-	n, err := p.r.Read(b)
-	if n > 0 {
-		p.sent += int64(n)
-		p.report(p.sent)
-	}
-	return n, err
-}
-
 // NewClient builds a sending client over the given dialer.
 func NewClient(cfg ClientConfig) (*Client, error) {
 	if cfg.Dial == nil {
@@ -301,6 +285,12 @@ func (c *Client) uploadFile(ctx context.Context, base string, id OfferID, index
 			if ctx.Err() != nil {
 				return ctx.Err()
 			}
+			// A refusal and a vanished offer are answers, not transport
+			// hiccups: the receiver has withdrawn, and retrying only keeps
+			// pushing bytes at someone who said no.
+			if errors.Is(err, ErrNotAccepted) || errors.Is(err, ErrOfferNotFound) {
+				return err
+			}
 			lastErr = err
 			log.Debugf("upload attempt %d for %s: %v", attempt+1, p.Meta.Name, err)
 			continue
@@ -355,6 +345,9 @@ func (c *Client) putFile(ctx context.Context, base string, id OfferID, index int
 	if resp.StatusCode == http.StatusForbidden {
 		return ErrNotAccepted
 	}
+	if resp.StatusCode == http.StatusNotFound {
+		return ErrOfferNotFound
+	}
 	if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK {
 		return statusError(resp)
 	}

client/internal/filedrop/manager.go

@@ -18,12 +18,14 @@ import (
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
 )
 
-// The event kinds. Progress is not an event: live transfers are polled.
+// The event kinds. Progress is one of them, thinned to a few reports a second
+// by the reader that raises it; see progressInterval.
 const (
 	EventOffer EventKind = iota
 	EventCompleted
 	EventFailed
 	EventWithdrawn
+	EventProgress
 )
 
 // portSignalGrace bounds how long a failed attempt waits for one signal message
@@ -304,7 +306,7 @@ func (m *Manager) Cancel(id OfferID) {
 	}
 
 	if server != nil {
-		if offer, ok := server.Offers().Decide(id, DecisionDeclined); ok {
+		if offer, ok := server.Offers().Revoke(id); ok {
 			server.Spool().Remove(offer.ID)
 		}
 	}
@@ -409,6 +411,7 @@ func (m *Manager) runSend(ctx context.Context, client *Client, handle *sendHandl
 			total += n
 		}
 		m.history.SetProgress(transfer.ID, total)
+		m.emit(EventProgress, m.transferOf(transfer.ID))
 	}
 
 	if err := client.Upload(ctx, addr, remoteID, payloads, progress); err != nil {
@@ -484,7 +487,10 @@ func (m *Manager) failSend(ctx context.Context, id OfferID, err error) {
 
 	state := StateFailed
 	switch {
-	case errors.Is(err, ErrDeclined):
+	// Withdrawn mid-transfer reads the same way as refused up front: the
+	// receiver said no, which is an answer rather than a failure.
+	case errors.Is(err, ErrDeclined), errors.Is(err, ErrNotAccepted),
+		errors.Is(err, ErrOfferNotFound):
 		state = StateDeclined
 	case errors.Is(err, ErrExpired):
 		state = StateExpired
@@ -554,6 +560,7 @@ func (m *Manager) OnProgress(offer Offer, index int, received int64) {
 		total += n
 	}
 	m.history.SetProgress(offer.ID, total)
+	m.emit(EventProgress, m.transferOf(offer.ID))
 }
 
 // OnCompleted implements Notifier.

client/internal/filedrop/offer.go

@@ -121,6 +121,30 @@ func (s *OfferStore) Decide(id OfferID, decision Decision) (Offer, bool) {
 	return entry.offer.clone(), true
 }
 
+// Revoke withdraws consent for an offer whatever it has already answered, so an
+// upload in flight can be stopped. Decide only moves an offer out of Pending: a
+// transfer that is already running has no way back through it.
+func (s *OfferStore) Revoke(id OfferID) (Offer, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	entry, ok := s.offers[id]
+	if !ok {
+		return Offer{}, false
+	}
+
+	// An offer still waiting has a reader parked on its channel; one that was
+	// answered has already had it closed, and closing twice would panic.
+	if entry.offer.Decision == DecisionPending {
+		close(entry.decided)
+	}
+
+	entry.offer.Decision = DecisionDeclined
+	entry.offer.State = StateCancelled
+
+	return entry.offer.clone(), true
+}
+
 // Await blocks until a decision, expiry, or ctx cancellation.
 func (s *OfferStore) Await(ctx context.Context, sender PeerKey, id OfferID) (Offer, error) {
 	s.mu.RLock()

client/internal/filedrop/progress.go

@@ -0,0 +1,57 @@
+package filedrop
+
+import (
+	"io"
+	"time"
+)
+
+// progressInterval is the floor between two progress reports of one item. A
+// percent of a large file can pass in a millisecond, and every report ends up
+// writing history and waking the UI, so the rate is capped regardless.
+const progressInterval = 200 * time.Millisecond
+
+// progressReader reports how much of an item has moved as it is read. Reports
+// are thinned to one per whole percent, and to one per progressInterval when
+// the bytes outrun even that; the last one always goes through so a transfer
+// never stops a tick short of done.
+type progressReader struct {
+	r       io.Reader
+	sent    int64
+	total   int64
+	report  func(sent int64)
+	percent int
+	last    time.Time
+}
+
+func (p *progressReader) Read(b []byte) (int, error) {
+	n, err := p.r.Read(b)
+	if n == 0 {
+		return n, err
+	}
+
+	p.sent += int64(n)
+	if p.due(time.Now()) {
+		p.report(p.sent)
+	}
+	return n, err
+}
+
+// due decides whether the reader has moved enough, and waited long enough, to
+// be worth another report.
+func (p *progressReader) due(now time.Time) bool {
+	if p.total <= 0 || p.sent >= p.total {
+		return true
+	}
+
+	percent := int(p.sent * 100 / p.total)
+	if percent == p.percent {
+		return false
+	}
+	if !p.last.IsZero() && now.Sub(p.last) < progressInterval {
+		return false
+	}
+
+	p.percent = percent
+	p.last = now
+	return true
+}

client/internal/filedrop/progress_test.go

@@ -0,0 +1,108 @@
+package filedrop
+
+import (
+	"bytes"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProgressReader_ReportsEveryPercentAtMost(t *testing.T) {
+	var reports []int64
+	// Read one byte at a time out of a 1000 byte payload: without thinning that
+	// would be 1000 reports for 100 percent worth of progress.
+	reader := &progressReader{
+		r:      bytes.NewReader(make([]byte, 1000)),
+		total:  1000,
+		report: func(sent int64) { reports = append(reports, sent) },
+	}
+
+	buf := make([]byte, 1)
+	for {
+		_, err := reader.Read(buf)
+		if err == io.EOF {
+			break
+		}
+		require.NoError(t, err)
+	}
+
+	assert.LessOrEqual(t, len(reports), 100, "One report per percent at most")
+	assert.Equal(t, int64(1000), reports[len(reports)-1], "The finished item must be reported")
+}
+
+func TestProgressReader_HoldsBackFasterThanTheInterval(t *testing.T) {
+	reader := &progressReader{total: 1000, report: func(int64) {}}
+	start := time.Now()
+
+	reader.sent = 10
+	require.True(t, reader.due(start), "The first percent goes through")
+
+	reader.sent = 20
+	assert.False(t, reader.due(start.Add(progressInterval/2)),
+		"A percent arriving inside the interval waits")
+
+	assert.True(t, reader.due(start.Add(progressInterval)),
+		"The same percent goes through once the interval has passed")
+}
+
+func TestProgressReader_ReportsTheLastByteWhateverTheInterval(t *testing.T) {
+	reader := &progressReader{total: 1000, report: func(int64) {}}
+	start := time.Now()
+
+	reader.sent = 500
+	require.True(t, reader.due(start))
+
+	reader.sent = 1000
+	assert.True(t, reader.due(start.Add(time.Millisecond)),
+		"A finished item is never held back by the interval")
+}
+
+func TestProgressReader_ReportsEveryReadWithoutASize(t *testing.T) {
+	// A payload of unknown size has no percent to thin against, so every read
+	// is worth reporting rather than none.
+	reader := &progressReader{total: 0, report: func(int64) {}}
+
+	reader.sent = 1
+	assert.True(t, reader.due(time.Now()))
+}
+
+func TestAcceptedReader_StopsOnceTheOfferIsWithdrawn(t *testing.T) {
+	accepted := true
+	reader := &acceptedReader{
+		r:        bytes.NewReader(make([]byte, 1024)),
+		accepted: func() bool { return accepted },
+		// Backdated so the very first read checks rather than waiting out the
+		// interval.
+		last: time.Now().Add(-progressInterval),
+	}
+
+	buf := make([]byte, 128)
+	n, err := reader.Read(buf)
+	require.NoError(t, err)
+	assert.Equal(t, 128, n, "An accepted offer reads normally")
+
+	accepted = false
+	reader.last = time.Now().Add(-progressInterval)
+
+	_, err = reader.Read(buf)
+	assert.ErrorIs(t, err, ErrNotAccepted, "A withdrawn offer stops the copy")
+}
+
+func TestAcceptedReader_ChecksNoMoreOftenThanTheInterval(t *testing.T) {
+	checks := 0
+	reader := &acceptedReader{
+		r:        bytes.NewReader(make([]byte, 4096)),
+		accepted: func() bool { checks++; return true },
+	}
+
+	buf := make([]byte, 1)
+	for range 32 {
+		_, err := reader.Read(buf)
+		require.NoError(t, err)
+	}
+
+	assert.LessOrEqual(t, checks, 1, "The offer store is not locked once per read")
+}

client/internal/filedrop/receiver.go

@@ -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)