diff --git a/client/internal/engine_filedrop.go b/client/internal/engine_filedrop.go index 39f6baa5b..a7ebd698c 100644 --- a/client/internal/engine_filedrop.go +++ b/client/internal/engine_filedrop.go @@ -32,9 +32,15 @@ func (e *Engine) startFileDrop() { if e.fileDrop == nil || e.fileDropRunning || e.wgInterface == nil { return } + e.setFileDropTunnel() + e.fileDrop.SetReceivingChangeHandler(e.onFileDropPolicyChange) + if e.config.BlockInbound { log.Info("file drop receiver is disabled because inbound connections are blocked") - e.setFileDropTunnel() + return + } + if !e.fileDrop.ReceivingEnabled() { + log.Info("file drop receiver is not started because receiving is turned off") return } @@ -70,7 +76,6 @@ func (e *Engine) startFileDrop() { e.setupFileDropPortRedirection(bound) - e.setFileDropTunnel() e.fileDropRunning = true } @@ -152,20 +157,62 @@ func (e *Engine) restartFileDrop() error { return nil } +// onFileDropPolicyChange binds or unbinds the receiver after the profile turned +// receiving on or off. The work is handed to a goroutine because it needs +// syncMsgMux, which the caller (a settings RPC) must not wait on and which the +// engine may itself hold while calling into the manager. +func (e *Engine) onFileDropPolicyChange() { + if e.ctx.Err() != nil { + return + } + + e.shutdownWg.Add(1) + go func() { + defer e.shutdownWg.Done() + + e.syncMsgMux.Lock() + defer e.syncMsgMux.Unlock() + + if e.fileDrop == nil || e.ctx.Err() != nil { + return + } + if e.fileDrop.ReceivingEnabled() { + e.startFileDrop() + return + } + if e.fileDropRunning { + e.unbindFileDropReceiver() + if err := e.fileDrop.DisableReceiving(); err != nil { + log.Warnf("failed to stop file drop receiver: %v", err) + } + e.fileDropRunning = false + e.fileDropPort = 0 + } + }() +} + +// unbindFileDropReceiver releases what the bind claimed outside the receiver +// itself. The caller must hold syncMsgMux. +func (e *Engine) unbindFileDropReceiver() { + if netstackNet := e.wgInterface.GetNet(); netstackNet != nil { + if registrar, ok := e.firewall.(interface { + UnregisterNetstackService(protocol nftypes.Protocol, port uint16) + }); ok { + registrar.UnregisterNetstackService(nftypes.TCP, e.fileDropPort) + } + } + e.removeFileDropPortRedirection(e.fileDropPort) +} + func (e *Engine) stopFileDrop() { if e.fileDrop == nil { return } + e.fileDrop.SetReceivingChangeHandler(nil) + e.fileDrop.ClearTunnel() if e.fileDropRunning { - if netstackNet := e.wgInterface.GetNet(); netstackNet != nil { - if registrar, ok := e.firewall.(interface { - UnregisterNetstackService(protocol nftypes.Protocol, port uint16) - }); ok { - registrar.UnregisterNetstackService(nftypes.TCP, e.fileDropPort) - } - } - e.removeFileDropPortRedirection(e.fileDropPort) + e.unbindFileDropReceiver() } if err := e.fileDrop.StopReceiver(); err != nil { diff --git a/client/internal/filedrop/manager.go b/client/internal/filedrop/manager.go index eec8f6a55..71725a3e6 100644 --- a/client/internal/filedrop/manager.go +++ b/client/internal/filedrop/manager.go @@ -197,12 +197,13 @@ func (m *Manager) AddReceiverListener(ctx context.Context, addr netip.AddrPort) return server.AddListener(ctx, addr) } -// StopReceiver shuts the receiving server down and drops the tunnel dialer. +// StopReceiver shuts the receiving server down, releasing its listeners and +// offer store. The spool is left alone: an interrupted transfer resumes from +// what is already staged once the receiver comes back. func (m *Manager) StopReceiver() error { m.mu.Lock() server := m.server m.server = nil - m.dial = nil m.mu.Unlock() if server == nil { @@ -211,6 +212,48 @@ func (m *Manager) StopReceiver() error { return server.Stop() } +// DisableReceiving stops the receiver because the profile turned receiving off. +// Unlike StopReceiver it discards what was in flight: nothing is coming back to +// claim the staged bytes, so the spool is emptied and every unfinished incoming +// transfer is settled as cancelled. +func (m *Manager) DisableReceiving() error { + err := m.StopReceiver() + + spool, serr := NewSpool(filepath.Join(m.dataDir, "spool")) + if serr != nil { + log.Warnf("failed to open file drop spool for cleanup: %v", serr) + } else { + spool.Purge() + } + + for _, t := range m.history.List() { + if t.Direction != DirectionReceived || t.terminal() { + continue + } + m.finishTransfer(t.ID, StateCancelled, "") + } + return err +} + +// ReceivingEnabled reports whether the profile accepts incoming offers. +func (m *Manager) ReceivingEnabled() bool { + return m.policy.Receiving() +} + +// SetReceivingChangeHandler registers a handler invoked when receiving is +// turned on or off, so the owner of the receiver can bind or unbind it. +func (m *Manager) SetReceivingChangeHandler(h func()) { + m.policy.SetChangeHandler(h) +} + +// ClearTunnel drops the tunnel dialer, leaving the manager unable to send. +func (m *Manager) ClearTunnel() { + m.mu.Lock() + m.dial = nil + m.senderName = "" + m.mu.Unlock() +} + // SetTunnel gives the manager the tunnel dialer and the local sender name. func (m *Manager) SetTunnel(dial DialFunc, senderName string) { m.mu.Lock() @@ -222,6 +265,7 @@ func (m *Manager) SetTunnel(dial DialFunc, senderName string) { // Close stops the receiver and aborts every outgoing transfer. func (m *Manager) Close() error { err := m.StopReceiver() + m.ClearTunnel() m.mu.Lock() for _, h := range m.sends { @@ -389,18 +433,18 @@ func (m *Manager) runSend(ctx context.Context, client *Client, handle *sendHandl addr := netip.AddrPortFrom(handle.ip, Port) remoteID, decision, err := client.Offer(ctx, addr, payloads) if err != nil { - m.failSend(ctx, transfer.ID, err) + m.failSend(ctx, transfer.ID, "offer", err) return } m.storeRemote(handle, addr, remoteID) decision, err = client.AwaitDecision(ctx, addr, remoteID, decision) if err != nil { - m.failSend(ctx, transfer.ID, err) + m.failSend(ctx, transfer.ID, "await decision", err) return } if err := decisionError(decision); err != nil { - m.failSend(ctx, transfer.ID, err) + m.failSend(ctx, transfer.ID, "decision", err) return } @@ -418,7 +462,7 @@ func (m *Manager) runSend(ctx context.Context, client *Client, handle *sendHandl } if err := client.Upload(ctx, addr, remoteID, payloads, progress); err != nil { - m.failSend(ctx, transfer.ID, err) + m.failSend(ctx, transfer.ID, "upload", err) return } @@ -434,8 +478,14 @@ func (m *Manager) storeRemote(handle *sendHandle, addr netip.AddrPort, remoteID handle.remoteID = remoteID } -func (m *Manager) failSend(ctx context.Context, id OfferID, err error) { +// failSend settles an outgoing transfer that did not complete. The step that +// broke is logged here, with the peer, so a failure can be read from the client +// log alone rather than only from the event the UI shows. +func (m *Manager) failSend(ctx context.Context, id OfferID, step string, err error) { + transfer := m.transferOf(id) + if ctx.Err() != nil { + log.Debugf("file drop send %s to %s cancelled during %s", id, transfer.PeerName, step) m.finishTransfer(id, StateCancelled, "") return } @@ -459,6 +509,21 @@ func (m *Manager) failSend(ctx context.Context, id OfferID, err error) { reason = ReasonUnreachable } } + + switch state { + case StateFailed: + if reason == ReasonUnreachable { + log.Warnf("file drop send %s to %s (%s): peer unreachable during %s: %v", + id, transfer.PeerName, transfer.PeerKey, step, err) + } else { + log.Warnf("file drop send %s to %s (%s) failed during %s: %v", + id, transfer.PeerName, transfer.PeerKey, step, err) + } + default: + log.Debugf("file drop send %s to %s ended as %s during %s: %v", + id, transfer.PeerName, state, step, err) + } + m.finishTransferReason(id, state, message, reason) m.emit(EventFailed, m.transferOf(id)) } @@ -554,10 +619,13 @@ func (m *Manager) OnCompleted(offer Offer) { // OnFailed implements Notifier. func (m *Manager) OnFailed(offer Offer, err error) { if errors.Is(err, ErrExpired) { + log.Debugf("file drop offer %s from %s expired unanswered", offer.ID, offer.SenderName) m.finishTransfer(offer.ID, StateExpired, "") m.emit(EventWithdrawn, m.transferOf(offer.ID)) return } + log.Warnf("file drop receive %s from %s (%s) failed: %v", + offer.ID, offer.SenderName, offer.Sender, err) m.finishTransfer(offer.ID, StateFailed, err.Error()) m.emit(EventFailed, m.transferOf(offer.ID)) } diff --git a/client/internal/filedrop/policy.go b/client/internal/filedrop/policy.go index dd0920dc0..bd7e82264 100644 --- a/client/internal/filedrop/policy.go +++ b/client/internal/filedrop/policy.go @@ -28,10 +28,11 @@ type Policy struct { // PolicyStore holds the receiving policy of one profile and evaluates it per sender. type PolicyStore struct { - mu sync.RWMutex - profile profilemanager.ID - policy Policy - store Store + mu sync.RWMutex + profile profilemanager.ID + policy Policy + store Store + onChange func() } // NewPolicyStore returns an in-memory store seeded with the default policy. @@ -107,6 +108,22 @@ func (s *PolicyStore) Profile() profilemanager.ID { return s.profile } +// SetChangeHandler registers a handler invoked when the base mode changes, so +// the owner of the receiver can bind or unbind it. It runs on the goroutine +// that made the change, with no store lock held. +func (s *PolicyStore) SetChangeHandler(h func()) { + s.mu.Lock() + s.onChange = h + s.mu.Unlock() +} + +// Receiving reports whether the profile accepts incoming offers at all. +func (s *PolicyStore) Receiving() bool { + s.mu.RLock() + defer s.mu.RUnlock() + return s.policy.Mode != ModeOff +} + // Get returns a copy of the current policy. func (s *PolicyStore) Get() Policy { s.mu.RLock() @@ -121,11 +138,16 @@ func (s *PolicyStore) Set(p Policy) error { } s.mu.Lock() + was := s.policy.Mode s.policy = p.normalized() - store, stored := s.store, s.policy.clone() + store, stored, changed, notify := s.store, s.policy.clone(), was != s.policy.Mode, s.onChange s.mu.Unlock() - return saveSection(store, namespacePolicy, stored) + err := saveSection(store, namespacePolicy, stored) + if changed && notify != nil { + notify() + } + return err } // SetMode changes the base mode, leaving per-sender rules untouched. @@ -135,11 +157,16 @@ func (s *PolicyStore) SetMode(m Mode) error { } s.mu.Lock() + changed := s.policy.Mode != m s.policy.Mode = m - store, stored := s.store, s.policy.clone() + store, stored, notify := s.store, s.policy.clone(), s.onChange s.mu.Unlock() - return saveSection(store, namespacePolicy, stored) + err := saveSection(store, namespacePolicy, stored) + if changed && notify != nil { + notify() + } + return err } // SetSenderRule sets or clears the override for a single sender. diff --git a/client/internal/filedrop/receiver.go b/client/internal/filedrop/receiver.go index 45e17fdee..926a65752 100644 --- a/client/internal/filedrop/receiver.go +++ b/client/internal/filedrop/receiver.go @@ -176,7 +176,8 @@ func (r *receiver) upload(sender senderIdentity, id OfferID, index int, offset i if err != nil { r.offers.SetState(id, StateFailed) r.notifyFailed(offer, err) - log.Debugf("stage payload for offer %s file %d: %v", id, index, err) + log.Warnf("file drop receive %s from %s: stage payload %d (%s): %v", + id, offer.SenderName, index, offer.Files[index].Name, err) return fmt.Errorf("%w: stage payload", ErrStorage) } diff --git a/client/internal/filedrop/spool.go b/client/internal/filedrop/spool.go index 91ea53a5b..d0d29981c 100644 --- a/client/internal/filedrop/spool.go +++ b/client/internal/filedrop/spool.go @@ -112,6 +112,20 @@ func (s *Spool) Remove(id OfferID) { } } +// Purge removes every staged payload. +func (s *Spool) Purge() { + entries, err := os.ReadDir(s.root) + if err != nil { + log.Debugf("read spool root: %v", err) + return + } + for _, entry := range entries { + if err := os.RemoveAll(filepath.Join(s.root, entry.Name())); err != nil { + log.Debugf("remove spool entry: %v", err) + } + } +} + // Cleanup removes offer directories older than maxAge. func (s *Spool) Cleanup(maxAge time.Duration, now time.Time) { entries, err := os.ReadDir(s.root)