diff --git a/client/android/filedrop_transfer.go b/client/android/filedrop_transfer.go index 5fef54a67..cba686ec3 100644 --- a/client/android/filedrop_transfer.go +++ b/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 diff --git a/client/internal/filedrop/client.go b/client/internal/filedrop/client.go index 2bb909089..bfaa8b046 100644 --- a/client/internal/filedrop/client.go +++ b/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) } diff --git a/client/internal/filedrop/manager.go b/client/internal/filedrop/manager.go index b3ec25cd8..ae1bfbddb 100644 --- a/client/internal/filedrop/manager.go +++ b/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. diff --git a/client/internal/filedrop/offer.go b/client/internal/filedrop/offer.go index 0f174f15e..3f28294ae 100644 --- a/client/internal/filedrop/offer.go +++ b/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() diff --git a/client/internal/filedrop/progress.go b/client/internal/filedrop/progress.go new file mode 100644 index 000000000..a29d4cc6d --- /dev/null +++ b/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 +} diff --git a/client/internal/filedrop/progress_test.go b/client/internal/filedrop/progress_test.go new file mode 100644 index 000000000..8b35de347 --- /dev/null +++ b/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") +} diff --git a/client/internal/filedrop/receiver.go b/client/internal/filedrop/receiver.go index d787e4ecf..3f5b610ea 100644 --- a/client/internal/filedrop/receiver.go +++ b/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)