relay: a reply that never comes fails after its timeout, not the sync's

internal/p2p/transport_wan.go

@@ -38,6 +38,9 @@ func (t *wanTransport) FetchManifest(ctx context.Context, peer syncengine.Peer,
 		Name: q.Name, SavePath: q.SavePath, IsFile: q.IsFile,
 		AppID: q.AppID, CoverURL: q.CoverURL,
 	}
+	// The peer holds its answer while a sync is writing the save, for up to
+	// ServeSettleWait, before it has the usual time to send it.
+	ctx = withReplyWait(ctx, wanRequestTimeout+syncengine.ServeSettleWait)
 	raw, err := t.wan.Request(ctx, peer.ID, "/manifest/"+gameID, "GET", body)
 	if err != nil {
 		return syncengine.ManifestResponse{}, err
@@ -61,14 +64,11 @@ func (t *wanTransport) FetchManifest(ctx context.Context, peer syncengine.Peer,
 const slowLinkBytesPerSec = 128 << 10
 
 func (t *wanTransport) FetchBlocks(ctx context.Context, peer syncengine.Peer, ref syncengine.FileRef, blockIndices []int, blockSize int) ([]syncengine.BlockData, error) {
-	// Give the request a deadline proportional to what it's actually moving,
-	// so a big batch on a poor connection isn't cut off mid-flight.
-	if _, hasDeadline := ctx.Deadline(); !hasDeadline && blockSize > 0 {
+	// Give the reply time proportional to what it's actually moving, so a big
+	// batch on a poor connection isn't cut off mid-flight.
+	if blockSize > 0 {
 		expected := int64(len(blockIndices)) * int64(blockSize)
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithTimeout(ctx,
-			wanRequestTimeout+time.Duration(expected/slowLinkBytesPerSec)*time.Second)
-		defer cancel()
+		ctx = withReplyWait(ctx, wanRequestTimeout+time.Duration(expected/slowLinkBytesPerSec)*time.Second)
 	}
 
 	raw, err := t.wan.Request(ctx, peer.ID, "/blocks/"+ref.GameID, "POST", map[string]any{

internal/p2p/wanclient.go

@@ -25,6 +25,11 @@ const (
 	wanRequestTimeout    = 30 * time.Second
 	wanPeerExpiry        = 20 * time.Second
 
+	// wanPresenceWriteEvery is how often a peer that stays online has its
+	// last-seen time written. Well inside wanPeerExpiry, so a peer that keeps
+	// sending is never expired between two writes.
+	wanPresenceWriteEvery = 5 * time.Second
+
 	// wanReconnectMaxDelay caps the exponential backoff. A relay that is
 	// genuinely down must not be hammered every 5s forever, and when it is
 	// merely cycling, every client in every room would otherwise pile onto
@@ -625,6 +630,32 @@ func (w *WanClient) Request(ctx context.Context, peerID, route, method string, b
 	return nil, fmt.Errorf("%s", errBody.Error)
 }
 
+// replyWaitKey carries how long a request may wait for its reply; see
+// withReplyWait.
+type replyWaitKey struct{}
+
+// withReplyWait lets the requests made with ctx wait up to d for their reply
+// instead of wanRequestTimeout: for an answer that legitimately takes longer,
+// such as a payload sized to the save data it moves, or a peer that holds its
+// answer while a sync is writing the save.
+func withReplyWait(ctx context.Context, d time.Duration) context.Context {
+	return context.WithValue(ctx, replyWaitKey{}, d)
+}
+
+// replyWait is how long a request waits for its reply: wanRequestTimeout, or
+// longer when the caller asked for it with withReplyWait — and never the
+// caller's whole deadline. The relay drops frames for a client that can't keep
+// up (relay/server.go), and a sync runs under a 30-minute deadline: a request
+// or reply that was dropped then stalled the sync for half an hour instead of
+// failing it after the timeout, to be retried.
+func replyWait(ctx context.Context) time.Duration {
+	timeout := wanRequestTimeout
+	if d, ok := ctx.Value(replyWaitKey{}).(time.Duration); ok && d > timeout {
+		timeout = d
+	}
+	return timeout
+}
+
 // exchange sends one request and waits for its reply. err is for a reply
 // that never came — the relay offline, a timeout — and a reply of any status
 // is returned as one, so a caller can tell "refused" from "not delivered".
@@ -657,16 +688,7 @@ func (w *WanClient) exchange(ctx context.Context, peerID, route, method string,
 
 	w.send(w.buildRequest(peerID, msgID, route, method, rawBody, authKey))
 
-	// wanRequestTimeout is a floor, not a ceiling: a caller moving several
-	// megabytes of save data sets a deadline sized to the payload, and a flat
-	// 30s would abandon a transfer that was making perfectly good progress on
-	// a slow link.
-	timeout := wanRequestTimeout
-	if deadline, ok := ctx.Deadline(); ok {
-		if until := time.Until(deadline); until > timeout {
-			timeout = until
-		}
-	}
+	timeout := replyWait(ctx)
 
 	select {
 	case <-ctx.Done():

internal/p2p/wanclient_handlers.go

@@ -224,13 +224,24 @@ func (w *WanClient) trackPresence(msg RelayMessage) {
 	if err == nil {
 		wasOffline := peer.Status != "online"
 		changed := wasOffline || peer.Address != "relay"
+		now := time.Now().UnixMilli()
+		// Written when something changed, and otherwise only every few
+		// seconds, as the LAN does (routes.go). Every frame from the peer
+		// lands here — each block of a transfer, each heartbeat — on the one
+		// goroutine that reads the relay, and a synced write per frame held
+		// that reading up behind the disk. The relay drops frames for a
+		// client that falls behind, so a slow disk turned into lost replies.
+		due := changed || (msg.Port > 0 && msg.Port != peer.Port) ||
+			now-peer.LastSeenMs >= wanPresenceWriteEvery.Milliseconds()
 		peer.Status = "online"
 		peer.Address = "relay"
-		peer.LastSeenMs = time.Now().UnixMilli()
+		peer.LastSeenMs = now
 		if msg.Port > 0 {
 			peer.Port = msg.Port
 		}
-		_ = w.engine.Store.UpdatePeer(peer)
+		if due {
+			_ = w.engine.Store.UpdatePeer(peer)
+		}
 		if wasOffline {
 			w.engine.Log("info", fmt.Sprintf("peer %q came online via WAN; auto-syncing", peer.Name))
 			w.engine.GoSync(func(ctx context.Context) { w.engine.SyncAllGames(ctx) })

internal/p2p/wanclient_replywait_test.go

@@ -0,0 +1,73 @@
+package p2p
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// A request waits wanRequestTimeout for its reply, however long the caller's
+// own deadline: a sync runs under 30 minutes, and a request or reply the relay
+// dropped kept the sync waiting all of that, where it should have failed and
+// been retried.
+func TestReplyWait_NotTheCallersWholeDeadline(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+	if got := replyWait(ctx); got != wanRequestTimeout {
+		t.Errorf("under a 30-minute deadline a reply is waited for %v, want %v", got, wanRequestTimeout)
+	}
+	if got := replyWait(context.Background()); got != wanRequestTimeout {
+		t.Errorf("with no deadline a reply is waited for %v, want %v", got, wanRequestTimeout)
+	}
+}
+
+// A request whose answer legitimately takes longer asks for it, and gets it;
+// asking for less than the floor changes nothing.
+func TestReplyWait_LongerWhenAskedFor(t *testing.T) {
+	if got := replyWait(withReplyWait(context.Background(), 3*time.Minute)); got != 3*time.Minute {
+		t.Errorf("asked for 3m, waits %v", got)
+	}
+	if got := replyWait(withReplyWait(context.Background(), time.Second)); got != wanRequestTimeout {
+		t.Errorf("asked for 1s, waits %v; want the floor %v", got, wanRequestTimeout)
+	}
+}
+
+// Every frame from a paired peer is presence; only a change, or a last-seen
+// time going stale, is written. A write per frame held up the one goroutine
+// that reads the relay behind the disk.
+func TestTrackPresence_WritesOnlyWhenDue(t *testing.T) {
+	f := newRaceFixture(t, false)
+	peer, err := f.s.GetPeer(f.peer.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	peer.Status = "online"
+	peer.Address = "relay"
+	seen := time.Now().Add(-time.Second).UnixMilli()
+	peer.LastSeenMs = seen
+	if err := f.s.UpdatePeer(peer); err != nil {
+		t.Fatal(err)
+	}
+	w := &WanClient{engine: f.e, discovered: map[string]WanPeer{}}
+	ping := RelayMessage{Type: "ping", From: f.peer.ID, Port: peer.Port}
+
+	w.trackPresence(ping)
+	if got, _ := f.s.GetPeer(f.peer.ID); got.LastSeenMs != seen {
+		t.Errorf("a heartbeat right after the last write was written again (last seen %d → %d)", seen, got.LastSeenMs)
+	}
+
+	stale := time.Now().Add(-wanPresenceWriteEvery - time.Second).UnixMilli()
+	peer.LastSeenMs = stale
+	if err := f.s.UpdatePeer(peer); err != nil {
+		t.Fatal(err)
+	}
+	w.trackPresence(ping)
+	if got, _ := f.s.GetPeer(f.peer.ID); got.LastSeenMs <= stale {
+		t.Errorf("a stale last-seen time was not refreshed (%d)", got.LastSeenMs)
+	}
+
+	w.trackPresence(RelayMessage{Type: "ping", From: f.peer.ID, Port: peer.Port + 1})
+	if got, _ := f.s.GetPeer(f.peer.ID); got.Port != peer.Port+1 {
+		t.Errorf("a new port was not written at once: %d, want %d", got.Port, peer.Port+1)
+	}
+}