diff --git a/internal/p2p/transport_wan.go b/internal/p2p/transport_wan.go index 859e91ef..8ed1811b 100644 --- a/internal/p2p/transport_wan.go +++ b/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{ diff --git a/internal/p2p/wanclient.go b/internal/p2p/wanclient.go index 4c21f47c..da3df379 100644 --- a/internal/p2p/wanclient.go +++ b/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(): diff --git a/internal/p2p/wanclient_handlers.go b/internal/p2p/wanclient_handlers.go index 95203dd3..df1c61be 100644 --- a/internal/p2p/wanclient_handlers.go +++ b/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) }) diff --git a/internal/p2p/wanclient_replywait_test.go b/internal/p2p/wanclient_replywait_test.go new file mode 100644 index 00000000..d2b0785f --- /dev/null +++ b/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) + } +}