tracker/server.go | 28 ++++++++++++++++++++++++++-- tracker/upstream-announcing.go | 18 ++++++++++++++++++ diff --git a/tracker/server.go b/tracker/server.go index 0d5068d48bf2ffd19f08fd61c89039e8823acb88..c74cbd245f3c7617b404fd8ae5c197def27168bd 100644 --- a/tracker/server.go +++ b/tracker/server.go @@ -38,10 +38,12 @@ GetPeers(ctx context.Context, infoHash InfoHash, opts GetPeersOpts) ([]PeerInfo, error) } type AnnounceHandler struct { - AnnounceTracker AnnounceTracker + AnnounceTracker AnnounceTracker + UpstreamTrackers []Client UpstreamTrackerUrls []string UpstreamAnnouncePeerId [20]byte + UpstreamAnnounceGate UpstreamAnnounceGater mu sync.Mutex // Operations are only removed when all the upstream peers have been tracked. @@ -174,7 +176,8 @@ return } func (me *AnnounceHandler) augmentPeersFromUpstream(infoHash [20]byte) augmentationOperation { - announceCtx, cancel := context.WithTimeout(context.Background(), time.Minute) + const announceTimeout = time.Minute + announceCtx, cancel := context.WithTimeout(context.Background(), announceTimeout) subReq := AnnounceRequest{ InfoHash: infoHash, PeerId: me.UpstreamAnnouncePeerId, @@ -190,9 +193,30 @@ client := me.UpstreamTrackers[i] url := me.UpstreamTrackerUrls[i] pendingUpstreams.Add(1) go func() { + started, err := me.UpstreamAnnounceGate.Start(announceCtx, url, infoHash, announceTimeout) + if err != nil { + log.Printf("error reserving announce for %x to %v: %v", infoHash, url, err) + } + if err != nil || !started { + peersChan <- nil + return + } + log.Printf("announcing %x upstream to %v", infoHash, url) resp, err := client.Announce(announceCtx, subReq, AnnounceOpt{ UserAgent: "aragorn", }) + interval := resp.Interval + go func() { + if interval < 5*60 { + // This is as much to reduce load on upstream trackers in the event of errors, + // as it is to reduce load on our peer store. + interval = 5 * 60 + } + err := me.UpstreamAnnounceGate.Completed(context.Background(), url, infoHash, interval) + if err != nil { + log.Printf("error recording completed announce for %x to %v: %v", infoHash, url, err) + } + }() peersChan <- resp.Peers if err != nil { log.Levelf(log.Warning, "error announcing to upstream %q: %v", url, err) diff --git a/tracker/upstream-announcing.go b/tracker/upstream-announcing.go new file mode 100644 index 0000000000000000000000000000000000000000..ab5a5fb34d133e28fbab35a6c4cefb489d1b58cf --- /dev/null +++ b/tracker/upstream-announcing.go @@ -0,0 +1,18 @@ +package tracker + +import ( + "context" + "time" +) + +type UpstreamAnnounceGater interface { + Start(ctx context.Context, tracker string, infoHash InfoHash, + // How long the announce block remains before discarding it. + timeout time.Duration, + ) (bool, error) + Completed( + ctx context.Context, tracker string, infoHash InfoHash, + // Num of seconds reported by tracker, or some suitable value the caller has chosen. + interval int32, + ) error +}