| 80 | } |
| 81 | |
| 82 | func NewSyncServer(storageManager StorageManagerReader, db ethdb.Database, m SyncServerMetrics, lg log.Logger) *SyncServer { |
| 83 | // We should never allow over 1000 different peers to churn through quickly, |
| 84 | // so it's fine to prune rate-limit details past this. |
| 85 | |
| 86 | peerRateLimits, _ := simplelru.NewLRU[peer.ID, *peerStat](1000, nil) |
| 87 | // 3 sync requests per second, with 2 burst |
| 88 | globalRequestsRL := rate.NewLimiter(globalServerBlocksRateLimit, globalServerBlocksBurst) |
| 89 | |
| 90 | if m == nil { |
| 91 | m = metrics.NoopMetrics |
| 92 | } |
| 93 | var providedBlobs map[uint64]uint64 |
| 94 | if status, _ := db.Get(ProvidedBlobsKey); status != nil { |
| 95 | if err := json.Unmarshal(status, &providedBlobs); err != nil { |
| 96 | lg.Error("Failed to decode provided blobs", "err", err) |
| 97 | } |
| 98 | } |
| 99 | |
| 100 | server := SyncServer{ |
| 101 | lg: lg, |
| 102 | storageManager: storageManager, |
| 103 | db: db, |
| 104 | providedBlobs: make(map[uint64]uint64), |
| 105 | exitCh: make(chan struct{}), |
| 106 | metrics: m, |
| 107 | peerRateLimits: peerRateLimits, |
| 108 | globalRequestsRL: globalRequestsRL, |
| 109 | } |
| 110 | |
| 111 | for _, shardId := range storageManager.Shards() { |
| 112 | if providedBlobs != nil { |
| 113 | if blobs, ok := providedBlobs[shardId]; ok { |
| 114 | server.providedBlobs[shardId] = blobs |
| 115 | continue |
| 116 | } |
| 117 | } |
| 118 | server.providedBlobs[shardId] = 0 |
| 119 | } |
| 120 | go server.SaveProvidedBlobs() |
| 121 | return &server |
| 122 | } |
| 123 | |
| 124 | // HandleGetBlobsByRangeRequest is a stream handler function to register the L2 unsafe payloads alt-sync protocol. |
| 125 | // See MakeStreamHandler to transform this into a LibP2P handler function. |