1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
package requester
import (
"context"
"errors"
"time"
"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
)
const (
maxBidderRepos = 32
bidCatchUpDelay = 5 * time.Second
)
func consideredBidderDIDs(bidders []Bidder, only []string) []string {
dids := append([]string(nil), only...)
for _, bidder := range bidders {
dids = append(dids, bidder.DID)
}
dids = BidderDIDList(dids)
if len(dids) > maxBidderRepos {
dids = dids[:maxBidderRepos]
}
return dids
}
func (r *contractRunner) catchUpBids(ctx context.Context, dids []string, rfpURI, rfpCID string) {
if len(dids) == 0 || rfpURI == "" || rfpCID == "" {
return
}
delay := min(bidCatchUpDelay, max(r.q.cfg.BidWindow/2, time.Second))
r.q.wg.Go(func() {
timer := time.NewTimer(delay)
defer timer.Stop()
select {
case <-ctx.Done():
return
case <-timer.C:
}
for _, did := range dids {
record, err := r.q.repo.GetRecord(ctx, did, market.BidNSID, market.BidRKey(did, rfpURI, rfpCID))
if err != nil {
if !errors.Is(err, market.ErrNotFound) && ctx.Err() == nil {
r.log().Debug("requester_bid_catch_up_failed", "bidder", did, "err", err)
}
continue
}
if err := r.q.storeBidMailbox(ctx, did, record.URI, record.CID, record.Value); err != nil {
if ctx.Err() == nil {
r.log().Warn("requester_bid_catch_up_failed", "bidder", did, "bid", record.URI, "err", err)
}
continue
}
r.log().Info("requester_bid_catch_up", "bidder", did, "bid", record.URI)
}
})
}
