[client] Keep a withdrawal from discarding a payload being delivered

Completing an offer and delivering it ran with no lock held across the two,
while a sender could send DELETE at any point and withdraw ran spool.Remove
without looking at the offer's state. A withdrawal landing in that window
deleted the staged bytes out from under the copy: delivery failed with
"open spooled file: no such file or directory", the transfer was recorded as
failed, and the payload was gone although the sender had seen its upload
succeed.

A completed offer is no longer withdrawable, and publishing or discarding one
offer's payloads now serialises on a per-offer lock the two sinks share, so a
removal waits for a delivery in flight instead of racing it. deliver() drops
the spool as its last step and reaches it through removeLocked, since the
lock it would otherwise retake is the one it already holds.

client/internal/filedrop/delivery.go

@@ -20,7 +20,7 @@ func deliver(spool *Spool, offer Offer, destDir string) ([]string, error) {
 	}
 
 	if !files {
-		spool.Remove(offer.ID)
+		spool.removeLocked(offer.ID)
 		return nil, nil
 	}
 
@@ -47,7 +47,7 @@ func deliver(spool *Spool, offer Offer, destDir string) ([]string, error) {
 		delivered = append(delivered, dest)
 	}
 
-	spool.Remove(offer.ID)
+	spool.removeLocked(offer.ID)
 	return delivered, nil
 }
 

client/internal/filedrop/filedrop_test.go

@@ -307,6 +307,117 @@ func TestUploadIsBoundedByAnnouncedSize(t *testing.T) {
 	assert.Len(t, staged, int(announced), "staged size must be capped at the announced size")
 }
 
+func TestWithdrawIsRefusedOnceTheOfferCompleted(t *testing.T) {
+	spool, err := NewSpool(t.TempDir())
+	require.NoError(t, err)
+
+	store := NewOfferStore(time.Minute)
+	notifier := &recordingNotifier{}
+	r := &receiver{offers: store, policy: NewPolicyStore(testProfile), spool: spool, notifier: notifier}
+
+	content := "payload"
+	offer := store.Add(testPeer, "sender", []FileMeta{{Name: "a.txt", Size: int64(len(content))}}, DecisionAccepted)
+	require.NoError(t, spool.Prepare(offer.ID))
+	_, err = spool.Write(offer.ID, 0, "", 0, strings.NewReader(content), int64(len(content)))
+	require.NoError(t, err)
+	store.SetProgress(offer.ID, 0, int64(len(content)))
+
+	_, ok := store.Complete(offer.ID)
+	require.True(t, ok, "the offer completes once fully staged")
+
+	err = r.withdraw(senderIdentity{key: testPeer}, offer.ID)
+	require.ErrorIs(t, err, ErrNotAccepted, "a completed offer must not be withdrawn")
+
+	staged, serr := spool.Received(offer.ID, 0)
+	require.NoError(t, serr)
+	assert.Equal(t, int64(len(content)), staged, "the staged payload must survive the withdrawal")
+}
+
+func TestOfferLockSerialisesTheSameOfferOnly(t *testing.T) {
+	var locks offerLocks
+
+	release := locks.lock(OfferID("a"))
+
+	// A different offer must not be held up by it.
+	other := make(chan struct{})
+	go func() {
+		locks.lock(OfferID("b"))()
+		close(other)
+	}()
+	select {
+	case <-other:
+	case <-time.After(2 * time.Second):
+		t.Fatal("a second offer must not block on another offer's lock")
+	}
+
+	// The same offer must wait until the holder releases.
+	same := make(chan struct{})
+	go func() {
+		locks.lock(OfferID("a"))()
+		close(same)
+	}()
+	select {
+	case <-same:
+		t.Fatal("the same offer must not be entered while it is held")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	release()
+	select {
+	case <-same:
+	case <-time.After(2 * time.Second):
+		t.Fatal("releasing must let the waiter through")
+	}
+
+	locks.mu.Lock()
+	held := len(locks.locks)
+	locks.mu.Unlock()
+	assert.Zero(t, held, "released offers must not be retained in the lock map")
+}
+
+func TestDeliverHoldsTheOfferLockAgainstRemove(t *testing.T) {
+	spool, err := NewSpool(t.TempDir())
+	require.NoError(t, err)
+
+	store := NewOfferStore(time.Minute)
+	content := strings.Repeat("Z", 8<<10)
+	offer := store.Add(testPeer, "sender", []FileMeta{{Name: "a.bin", Size: int64(len(content))}}, DecisionAccepted)
+	require.NoError(t, spool.Prepare(offer.ID))
+	_, err = spool.Write(offer.ID, 0, "", 0, strings.NewReader(content), int64(len(content)))
+	require.NoError(t, err)
+
+	full, ok := store.Get(testPeer, offer.ID)
+	require.True(t, ok)
+
+	// Hold the offer's lock the way a running Deliver does, then let a
+	// withdrawal's Remove race it: the removal must wait rather than pull the
+	// staged bytes out from under the copy.
+	release := spool.lock(offer.ID)
+
+	removed := make(chan struct{})
+	go func() {
+		spool.Remove(offer.ID)
+		close(removed)
+	}()
+
+	select {
+	case <-removed:
+		t.Fatal("Remove must block while the offer lock is held")
+	case <-time.After(100 * time.Millisecond):
+	}
+	release()
+
+	delivered, derr := spool.Deliver(full, t.TempDir())
+	require.NoError(t, derr)
+	require.Len(t, delivered, 1)
+
+	info, serr := os.Stat(delivered[0])
+	require.NoError(t, serr)
+	assert.Equal(t, int64(len(content)), info.Size(), "the delivered payload must be whole")
+
+	<-removed
+}
+
 func TestCancelWithdrawsPendingOffer(t *testing.T) {
 	srv, client, notifier := startTestServer(t, ModeAsk, staticResolver{key: testPeer})
 

client/internal/filedrop/receiver.go

@@ -102,6 +102,13 @@ func (r *receiver) withdraw(sender senderIdentity, id OfferID) error {
 		return ErrOfferNotFound
 	}
 
+	// A completed offer has already been handed to delivery, which copies out
+	// of the spool: discarding it here would pull the staged bytes out from
+	// under that copy and lose a payload the sender was told had arrived.
+	if offer.State == StateCompleted {
+		return fmt.Errorf("%w: offer already completed", ErrNotAccepted)
+	}
+
 	r.offers.SetState(id, StateCancelled)
 	r.offers.Remove(id)
 	r.spool.Remove(id)

client/internal/filedrop/sink.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -50,8 +51,19 @@ type PlatformWriter interface {
 	Written() int64
 }
 
+// offerLocks serialises the calls that publish or discard one offer's staged
+// payloads. Delivery copies out of the spool while the sender may still send a
+// withdrawal, and the two arrive on different goroutines through different
+// types, so without this a DELETE landing mid-delivery removes the bytes the
+// copy is reading.
+type offerLocks struct {
+	mu    sync.Mutex
+	locks map[OfferID]*sync.Mutex
+}
+
 // platformSink adapts a PlatformSink to the Sink the receiver uses.
 type platformSink struct {
+	offerLocks
 	platform PlatformSink
 }
 
@@ -68,6 +80,34 @@ func NewPlatformSink(platform PlatformSink) (Sink, error) {
 	return &platformSink{platform: platform}, nil
 }
 
+// lock takes the lock guarding one offer's staged payloads and returns the
+// release. Entries are dropped on release, so the map tracks only offers with
+// a call in flight rather than growing with every offer ever seen.
+func (l *offerLocks) lock(id OfferID) func() {
+	l.mu.Lock()
+	if l.locks == nil {
+		l.locks = make(map[OfferID]*sync.Mutex)
+	}
+	m, ok := l.locks[id]
+	if !ok {
+		m = &sync.Mutex{}
+		l.locks[id] = m
+	}
+	l.mu.Unlock()
+
+	m.Lock()
+	return func() {
+		m.Unlock()
+
+		l.mu.Lock()
+		defer l.mu.Unlock()
+		if m.TryLock() {
+			m.Unlock()
+			delete(l.locks, id)
+		}
+	}
+}
+
 // DestinationLabel reports where the platform delivers payloads.
 func (s *platformSink) DestinationLabel() string {
 	return s.platform.DestinationLabel()
@@ -105,6 +145,8 @@ func (s *platformSink) Write(id OfferID, index int, name string, offset int64, r
 }
 
 func (s *platformSink) Deliver(offer Offer, _ string) ([]string, error) {
+	defer s.lock(offer.ID)()
+
 	delivered, err := s.platform.Deliver(string(offer.ID))
 	if err != nil {
 		return nil, err
@@ -113,6 +155,8 @@ func (s *platformSink) Deliver(offer Offer, _ string) ([]string, error) {
 }
 
 func (s *platformSink) Remove(id OfferID) {
+	defer s.lock(id)()
+
 	s.platform.Remove(string(id))
 }
 

client/internal/filedrop/spool.go

@@ -13,6 +13,7 @@ import (
 
 // Spool stages incoming payloads in an app-private directory.
 type Spool struct {
+	offerLocks
 	root string
 }
 
@@ -97,6 +98,8 @@ func (s *Spool) Write(id OfferID, index int, _ string, offset int64, r io.Reader
 // Deliver moves an offer's staged payloads into destDir under their announced
 // names and returns where each one landed.
 func (s *Spool) Deliver(offer Offer, destDir string) ([]string, error) {
+	defer s.lock(offer.ID)()
+
 	return deliver(s, offer, destDir)
 }
 
@@ -107,6 +110,15 @@ func (s *Spool) Path(id OfferID, index int) string {
 
 // Remove deletes an offer's staged payloads.
 func (s *Spool) Remove(id OfferID) {
+	defer s.lock(id)()
+
+	s.removeLocked(id)
+}
+
+// removeLocked discards one offer's payloads without taking its lock, for a
+// caller that already holds it. Deliver removes the spool as its last step and
+// would otherwise block on itself.
+func (s *Spool) removeLocked(id OfferID) {
 	if err := os.RemoveAll(s.OfferDir(id)); err != nil {
 		log.Debugf("remove spool dir: %v", err)
 	}