CVE-2023-25568: Allocation of Resources Without Limits or Throttling (CWE-770)
value: Boxo, formerly known as go-libipfs, is a library for building IPFS applications and implementations. In versions 0.4.0 and 0.5.0, if an attacker is able allocate arbitrary many bytes in the Bitswap server, those allocations are lasting even if the connection is closed. This affects users accepting untrusted connections with the Bitswap server and also affects users using the old API stubs at `github.com/ipfs/go-libipfs/bitswap` because users then transitively import `github.com/ipfs/go-libipfs/bitswap/server`. Boxo versions 0.6.0 and 0.4.1 contain a patch for this issue. As a workaround, those who are using the stub object at `github.com/ipfs/go-libipfs/bitswap` not taking advantage of the features provided by the server can refactor their code to use the new split API that will allow them to run in a client only mode: `github.com/ipfs/go-libipfs/bitswap/client`.
CWE description: The product allocates a reusable resource or group of resources on behalf of an actor without imposing any restrictions on the size or number of resources that can be allocated, in violation of the intended security policy for that actor.
CVSS v3 base score: 7.5
Published: 2023-05-10T14:15Z
Repository: https://github.com/ipfs/boxo
Language: Go
Files changed: bitswap/internal/defaults/defaults.go, bitswap/options.go, bitswap/server/internal/decision/engine.go, bitswap/server/internal/decision/engine_test.go, bitswap/server/internal/decision/peer_ledger.go, bitswap/server/server.go, examples/go.sum, go.mod, go.sum
Commit date: 2023-02-17 18:06:35 +0100
Commit message: bitswap/server/internal/decision: rewrite ledger inversion

Diff: diff --git a/bitswap/internal/defaults/defaults.go b/bitswap/internal/defaults/defaults.go
index 6f7c2e7..1b0de24 100644
--- a/bitswap/internal/defaults/defaults.go
+++ b/bitswap/internal/defaults/defaults.go
@@ -22,6 +22,9 @@ const (
 	// HasBlockBufferSize is the buffer size of the channel for new blocks
 	// that need to be provided. They should get pulled over by the
 	// provideCollector even before they are actually provided.
 	// TODO: Does this need to be this large givent that?
 	HasBlockBufferSize = 256
+
+	// Maximum size of the wantlist we are willing to keep in memory.
+	MaxQueuedWantlistEntiresPerPeer = 1024
 )
diff --git a/bitswap/options.go b/bitswap/options.go
index 4d5c4b4..1e2e090 100644
--- a/bitswap/options.go
+++ b/bitswap/options.go
@@ -27,10 +27,14 @@ func EngineTaskWorkerCount(count int) Option {
 
 func MaxOutstandingBytesPerPeer(count int) Option {
 	return Option{server.MaxOutstandingBytesPerPeer(count)}
 }
 
+func MaxQueuedWantlistEntriesPerPeer(count uint) Option {
+	return Option{server.MaxQueuedWantlistEntriesPerPeer(count)}
+}
+
 func TaskWorkerCount(count int) Option {
 	return Option{server.TaskWorkerCount(count)}
 }
 
 func ProvideEnabled(enabled bool) Option {
diff --git a/bitswap/server/internal/decision/engine.go b/bitswap/server/internal/decision/engine.go
index f21553f..29ac1aa 100644
--- a/bitswap/server/internal/decision/engine.go
+++ b/bitswap/server/internal/decision/engine.go
@@ -2,10 +2,11 @@
 package decision
 
 import (
 	"context"
 	"fmt"
+	"math/bits"
 	"sync"
 	"time"
 
 	"github.com/google/uuid"
 
@@ -145,13 +146,10 @@ type Engine struct {
 
 	tagQueued, tagUseful string
 
 	lock sync.RWMutex // protects the fields immediately below
 
-	// ledgerMap lists block-related Ledgers by their Partner key.
-	ledgerMap map[peer.ID]*ledger
-
 	// peerLedger saves which peers are waiting for a Cid
 	peerLedger *peerLedger
 
 	// an external ledger dealing with peer scores
 	scoreLedger ScoreLedger
@@ -185,10 +183,12 @@ type Engine struct {
 
 	peerBlockRequestFilter PeerBlockRequestFilter
 
 	bstoreWorkerCount          int
 	maxOutstandingBytesPerPeer int
+
+	maxQueuedWantlistEntriesPerPeer uint
 }
 
 // TaskInfo represents the details of a request from a peer.
 type TaskInfo struct {
 	Peer peer.ID
@@ -268,10 +268,19 @@ func WithMaxOutstandingBytesPerPeer(count int) Option {
 	return func(e *Engine) {
 		e.maxOutstandingBytesPerPeer = count
 	}
 }
 
+// WithMaxQueuedWantlistEntriesPerPeer limits how much individual entries each peer is allowed to send.
+// If a peer send us more than this we will truncate newest entries.
+// It defaults to DefaultMaxQueuedWantlistEntiresPerPeer.
+func WithMaxQueuedWantlistEntriesPerPeer(count uint) Option {
+	return func(e *Engine) {
+		e.maxQueuedWantlistEntriesPerPeer = count
+	}
+}
+
 func WithSetSendDontHave(send bool) Option {
 	return func(e *Engine) {
 		e.sendDontHaves = send
 	}
 }
@@ -328,11 +337,10 @@ func newEngine(
 	self peer.ID,
 	maxReplaceSize int,
 	opts ...Option,
 ) *Engine {
 	e := &Engine{
-		ledgerMap:                       make(map[peer.ID]*ledger),
 		scoreLedger:                     NewDefaultScoreLedger(),
 		bstoreWorkerCount:               defaults.BitswapEngineBlockstoreWorkerCount,
 		maxOutstandingBytesPerPeer:      defaults.BitswapMaxOutstandingBytesPerPeer,
 		peerTagger:                      peerTagger,
 		outbox:                          make(chan (<-chan *Envelope), outboxChanBuffer),
@@ -346,10 +354,11 @@ func newEngine(
 		pendingGauge:                    bmetrics.PendingEngineGauge(ctx),
 		activeGauge:                     bmetrics.ActiveEngineGauge(ctx),
 		targetMessageSize:               defaultTargetMessageSize,
 		tagQueued:                       fmt.Sprintf(tagFormat, "queued", uuid.New().String()),
 		tagUseful:                       fmt.Sprintf(tagFormat, "useful", uuid.New().String()),
+		maxQueuedWantlistEntriesPerPeer: defaults.MaxQueuedWantlistEntiresPerPeer,
 	}
 
 	for _, opt := range opts {
 		opt(e)
 	}
@@ -448,17 +457,14 @@ func (e *Engine) onPeerRemoved(p peer.ID) {
 	e.peerTagger.UntagPeer(p, e.tagQueued)
 }
 
 // WantlistForPeer returns the list of keys that the given peer has asked for
 func (e *Engine) WantlistForPeer(p peer.ID) []wl.Entry {
-	partner := e.findOrCreate(p)
-
-	partner.lk.Lock()
-	entries := partner.wantList.Entries()
-	partner.lk.Unlock()
+	e.lock.RLock()
+	defer e.lock.RUnlock()
 
-	return entries
+	return e.peerLedger.WantlistForPeer(p)
 }
 
 // LedgerForPeer returns aggregated data communication with a given peer.
 func (e *Engine) LedgerForPeer(p peer.ID) *Receipt {
 	return e.scoreLedger.GetReceipt(p)
@@ -603,16 +609,11 @@ func (e *Engine) Outbox() <-chan (<-chan *Envelope) {
 // Peers returns a slice of Peers with whom the local node has active sessions.
 func (e *Engine) Peers() []peer.ID {
 	e.lock.RLock()
 	defer e.lock.RUnlock()
 
-	response := make([]peer.ID, 0, len(e.ledgerMap))
-
-	for _, ledger := range e.ledgerMap {
-		response = append(response, ledger.Partner)
-	}
-	return response
+	return e.peerLedger.CollectPeerIDs()
 }
 
 // MessageReceived is called when a message is received from a remote peer.
 // For each item in the wantlist, add a want-have or want-block entry to the
 // request queue (this is later popped off by the workerTasks)
@@ -657,37 +658,38 @@ func (e *Engine) MessageReceived(ctx context.Context, p peer.ID, m bsmsg.BitSwap
 		log.Info("aborting message processing", err)
 		return
 	}
 
 	e.lock.Lock()
-	for _, entry := range wants {
-		e.peerLedger.Wants(p, entry.Cid)
-	}
-	for _, entry := range cancels {
-		e.peerLedger.CancelWant(p, entry.Cid)
-	}
-	e.lock.Unlock()
 
-	// Get the ledger for the peer
-	l := e.findOrCreate(p)
-	l.lk.Lock()
-	defer l.lk.Unlock()
-
-	// If the peer sent a full wantlist, replace the ledger's wantlist
 	if m.Full() {
-		l.wantList = wl.New()
+		e.peerLedger.ClearPeerWantlist(p)
 	}
 
-	var activeEntries []peertask.Task
+	s := uint(e.peerLedger.WantlistSizeForPeer(p))
+	if wouldBe := s + uint(len(wants)); wouldBe > e.maxQueuedWantlistEntriesPerPeer {
+		log.Debugw("wantlist overflow", "local", e.self, "remote", p, "would be", wouldBe)
+		// truncate wantlist to avoid overflow
+		available, o := bits.Sub(e.maxQueuedWantlistEntriesPerPeer, s, 0)
+		if o != 0 {
+			available = 0
+		}
+		wants = wants[:available]
+	}
 
-	// Remove cancelled blocks from the queue
+	for _, entry := range wants {
+		e.peerLedger.Wants(p, entry.Entry)
+	}
 	for _, entry := range cancels {
 		log.Debugw("Bitswap engine <- cancel", "local", e.self, "from", p, "cid", entry.Cid)
-		if l.CancelWant(entry.Cid) {
+		if e.peerLedger.CancelWant(p, entry.Cid) {
 			e.peerRequestQueue.Remove(entry.Cid, p)
 		}
 	}
+	e.lock.Unlock()
+
+	var activeEntries []peertask.Task
 
 	// Cancel a block operation
 	sendDontHave := func(entry bsmsg.Entry) {
 		// Only add the task to the queue if the requester wants a DONT_HAVE
 		if e.sendDontHaves && entry.SendDontHave {
@@ -722,13 +724,10 @@ func (e *Engine) MessageReceived(ctx context.Context, p peer.ID, m bsmsg.BitSwap
 	// For each want-have / want-block
 	for _, entry := range wants {
 		c := entry.Cid
 		blockSize, found := blockSizes[entry.Cid]
 
-		// Add each want-have / want-block to the ledger
-		l.Wants(c, entry.Priority, entry.WantType)
-
 		// If the block was not found
 		if !found {
 			log.Debugw("Bitswap engine: block not found", "local", e.self, "from", p, "cid", entry.Cid, "sendDontHave", entry.SendDontHave)
 			sendDontHave(entry)
 		} else {
@@ -761,11 +760,11 @@ func (e *Engine) MessageReceived(ctx context.Context, p peer.ID, m bsmsg.BitSwap
 		}
 	}
 
 	// Push entries onto the request queue
 	if len(activeEntries) > 0 {
-		e.peerRequestQueue.PushTasks(p, activeEntries...)
+		e.peerRequestQueue.PushTasksTruncated(e.maxQueuedWantlistEntriesPerPeer, p, activeEntries...)
 		e.updateMetrics()
 	}
 }
 
 // Split the want-have / want-block entries from the cancel entries
@@ -807,18 +806,14 @@ func (e *Engine) splitWantsDenials(p peer.ID, allWants []bsmsg.Entry) ([]bsmsg.E
 func (e *Engine) ReceivedBlocks(from peer.ID, blks []blocks.Block) {
 	if len(blks) == 0 {
 		return
 	}
 
-	l := e.findOrCreate(from)
-
 	// Record how many bytes were received in the ledger
-	l.lk.Lock()
-	defer l.lk.Unlock()
 	for _, blk := range blks {
 		log.Debugw("Bitswap engine <- block", "local", e.self, "from", from, "cid", blk.Cid(), "size", len(blk.RawData()))
-		e.scoreLedger.AddToReceivedBytes(l.Partner, len(blk.RawData()))
+		e.scoreLedger.AddToReceivedBytes(from, len(blk.RawData()))
 	}
 }
 
 // NotifyNewBlocks is called when new blocks becomes available locally, and in particular when the caller of bitswap
 // decide to store those blocks and make them available on the network.
@@ -833,50 +828,30 @@ func (e *Engine) NotifyNewBlocks(blks []blocks.Block) {
 		blockSizes[blk.Cid()] = len(blk.RawData())
 	}
 
 	// Check each peer to see if it wants one of the blocks we received
 	var work bool
-	missingWants := make(map[peer.ID][]cid.Cid)
 	for _, b := range blks {
 		k := b.Cid()
 
 		e.lock.RLock()
 		peers := e.peerLedger.Peers(k)
 		e.lock.RUnlock()
 
-		for _, p := range peers {
-			e.lock.RLock()
-			ledger, ok := e.ledgerMap[p]
-			e.lock.RUnlock()
-
-			if !ok {
-				// This can happen if the peer has disconnected while we're processing this list.
-				log.Debugw("failed to find peer in ledger", "peer", p)
-				missingWants[p] = append(missingWants[p], k)
-				continue
-			}
-			ledger.lk.RLock()
-			entry, ok := ledger.WantListContains(k)
-			ledger.lk.RUnlock()
-			if !ok {
-				// This can happen if the peer has canceled their want while we're processing this message.
-				log.Debugw("wantlist index doesn't match peer's wantlist", "peer", p)
-				missingWants[p] = append(missingWants[p], k)
-				continue
-			}
+		for _, entry := range peers {
 			work = true
 
 			blockSize := blockSizes[k]
 			isWantBlock := e.sendAsBlock(entry.WantType, blockSize)
 
 			entrySize := blockSize
 			if !isWantBlock {
 				entrySize = bsmsg.BlockPresenceSize(k)
 			}
 
-			e.peerRequestQueue.PushTasks(p, peertask.Task{
-				Topic:    entry.Cid,
+			e.peerRequestQueue.PushTasksTruncated(e.maxQueuedWantlistEntriesPerPeer, entry.Peer, peertask.Task{
+				Topic:    k,
 				Priority: int(entry.Priority),
 				Work:     entrySize,
 				Data: &taskData{
 					BlockSize:    blockSize,
 					HaveBlock:    true,
@@ -886,34 +861,10 @@ func (e *Engine) NotifyNewBlocks(blks []blocks.Block) {
 			})
 			e.updateMetrics()
 		}
 	}
 
-	// If we found missing wants (e.g., because the peer disconnected, we have some races here)
-	// remove them from the list. Unfortunately, we still have to re-check because the user
-	// could have re-connected in the meantime.
-	if len(missingWants) > 0 {
-		e.lock.Lock()
-		for p, wl := range missingWants {
-			if ledger, ok := e.ledgerMap[p]; ok {
-				ledger.lk.RLock()
-				for _, k := range wl {
-					if _, has := ledger.WantListContains(k); has {
-						continue
-					}
-					e.peerLedger.CancelWant(p, k)
-				}
-				ledger.lk.RUnlock()
-			} else {
-				for _, k := range wl {
-					e.peerLedger.CancelWant(p, k)
-				}
-			}
-		}
-		e.lock.Unlock()
-	}
-
 	if work {
 		e.signalNewWork()
 	}
 }
 
@@ -924,60 +875,45 @@ func (e *Engine) NotifyNewBlocks(blks []blocks.Block) {
 // send happen atomically
 
 // MessageSent is called when a message has successfully been sent out, to record
 // changes.
 func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage) {
-	l := e.findOrCreate(p)
-	l.lk.Lock()
-	defer l.lk.Unlock()
+	e.lock.Lock()
+	defer e.lock.Unlock()
 
 	// Remove sent blocks from the want list for the peer
 	for _, block := range m.Blocks() {
-		e.scoreLedger.AddToSentBytes(l.Partner, len(block.RawData()))
-		l.wantList.RemoveType(block.Cid(), pb.Message_Wantlist_Block)
+		e.scoreLedger.AddToSentBytes(p, len(block.RawData()))
+		e.peerLedger.CancelWantWithType(p, block.Cid(), pb.Message_Wantlist_Block)
 	}
 
 	// Remove sent block presences from the want list for the peer
 	for _, bp := range m.BlockPresences() {
 		// Don't record sent data. We reserve that for data blocks.
 		if bp.Type == pb.Message_Have {
-			l.wantList.RemoveType(bp.Cid, pb.Message_Wantlist_Have)
+			e.peerLedger.CancelWantWithType(p, bp.Cid, pb.Message_Wantlist_Have)
 		}
 	}
 }
 
 // PeerConnected is called when a new peer connects, meaning we should start
 // sending blocks.
 func (e *Engine) PeerConnected(p peer.ID) {
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
-	_, ok := e.ledgerMap[p]
-	if !ok {
-		e.ledgerMap[p] = newLedger(p)
-	}
-
 	e.scoreLedger.PeerConnected(p)
 }
 
 // PeerDisconnected is called when a peer disconnects.
 func (e *Engine) PeerDisconnected(p peer.ID) {
+	e.peerRequestQueue.Clear(p)
+
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
-	ledger, ok := e.ledgerMap[p]
-	if ok {
-		ledger.lk.RLock()
-		entries := ledger.Entries()
-		ledger.lk.RUnlock()
-
-		for _, entry := range entries {
-			e.peerLedger.CancelWant(p, entry.Cid)
-		}
-	}
-	delete(e.ledgerMap, p)
-
+	e.peerLedger.PeerDisconnected(p)
 	e.scoreLedger.PeerDisconnected(p)
 }
 
 // If the want is a want-have, and it's below a certain size, send the full
 // block (instead of sending a HAVE)
@@ -992,33 +928,10 @@ func (e *Engine) numBytesSentTo(p peer.ID) uint64 {
 
 func (e *Engine) numBytesReceivedFrom(p peer.ID) uint64 {
 	return e.LedgerForPeer(p).Recv
 }
 
-// ledger lazily instantiates a ledger
-func (e *Engine) findOrCreate(p peer.ID) *ledger {
-	// Take a read lock (as it's less expensive) to check if we have a ledger
-	// for the peer
-	e.lock.RLock()
-	l, ok := e.ledgerMap[p]
-	e.lock.RUnlock()
-	if ok {
-		return l
-	}
-
-	// There's no ledger, so take a write lock, then check again and create the
-	// ledger if necessary
-	e.lock.Lock()
-	defer e.lock.Unlock()
-	l, ok = e.ledgerMap[p]
-	if !ok {
-		l = newLedger(p)
-		e.ledgerMap[p] = l
-	}
-	return l
-}
-
 func (e *Engine) signalNewWork() {
 	// Signal task generation to restart (if stopped!)
 	select {
 	case e.workSignal <- struct{}{}:
 	default:
diff --git a/bitswap/server/internal/decision/engine_test.go b/bitswap/server/internal/decision/engine_test.go
index 35d35b1..9fd3b1b 100644
--- a/bitswap/server/internal/decision/engine_test.go
+++ b/bitswap/server/internal/decision/engine_test.go
@@ -144,31 +144,29 @@ func TestConsistentAccounting(t *testing.T) {
 	if receiver.Engine.numBytesSentTo(sender.Peer) != 0 || sender.Engine.numBytesReceivedFrom(receiver.Peer) != 0 {
 		t.Fatal("Bert didn't send bytes to Ernie")
 	}
 }
 
-func TestPeerIsAddedToPeersWhenMessageReceivedOrSent(t *testing.T) {
+func TestPeerIsAddedToPeersWhenMessageSent(t *testing.T) {
 	test.Flaky(t)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	sanfrancisco := newTestEngine(ctx, "sf")
 	seattle := newTestEngine(ctx, "sea")
 
 	m := message.New(true)
 
-	sanfrancisco.Engine.MessageSent(seattle.Peer, m)
+	// We need to request something for it to add us as partner.
+	m.AddEntry(blocks.NewBlock([]byte("Hæ")).Cid(), 0, pb.Message_Wantlist_Block, true)
+
 	seattle.Engine.MessageReceived(ctx, sanfrancisco.Peer, m)
 
 	if seattle.Peer == sanfrancisco.Peer {
 		t.Fatal("Sanity Check: Peers have same Key!")
 	}
 
-	if !peerIsPartner(seattle.Peer, sanfrancisco.Engine) {
-		t.Fatal("Peer wasn't added as a Partner")
-	}
-
 	if !peerIsPartner(sanfrancisco.Peer, seattle.Engine) {
 		t.Fatal("Peer wasn't added as a Partner")
 	}
 
 	seattle.Engine.PeerDisconnected(sanfrancisco.Peer)
@@ -1051,14 +1049,10 @@ func TestWantlistForPeer(t *testing.T) {
 
 	entries = e.WantlistForPeer(partner)
 	if len(entries) != 4 {
 		t.Fatal("expected wantlist to contain all wants from parter")
 	}
-	if entries[0].Priority != 4 || entries[1].Priority != 3 || entries[2].Priority != 2 || entries[3].Priority != 1 {
-		t.Fatal("expected wantlist to be sorted")
-	}
-
 }
 
 func TestTaskComparator(t *testing.T) {
 	test.Flaky(t)
 
diff --git a/bitswap/server/internal/decision/peer_ledger.go b/bitswap/server/internal/decision/peer_ledger.go
index c22322b..102dad4 100644
--- a/bitswap/server/internal/decision/peer_ledger.go
+++ b/bitswap/server/internal/decision/peer_ledger.go
@@ -1,46 +1,156 @@
 package decision
 
 import (
+	wl "github.com/ipfs/go-libipfs/bitswap/client/wantlist"
+	pb "github.com/ipfs/go-libipfs/bitswap/message/pb"
+
 	"github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p/core/peer"
 )
 
 type peerLedger struct {
-	cids map[cid.Cid]map[peer.ID]struct{}
+	// thoses two maps are inversions of each other
+	peers map[peer.ID]map[cid.Cid]entry
+	cids  map[cid.Cid]map[peer.ID]entry
 }
 
 func newPeerLedger() *peerLedger {
-	return &peerLedger{cids: make(map[cid.Cid]map[peer.ID]struct{})}
+	return &peerLedger{
+		peers: make(map[peer.ID]map[cid.Cid]entry),
+		cids:  make(map[cid.Cid]map[peer.ID]entry),
+	}
 }
 
-func (l *peerLedger) Wants(p peer.ID, k cid.Cid) {
-	m, ok := l.cids[k]
+func (l *peerLedger) Wants(p peer.ID, e wl.Entry) {
+	cids, ok := l.peers[p]
+	if !ok {
+		cids = make(map[cid.Cid]entry)
+		l.peers[p] = cids
+	}
+	cids[e.Cid] = entry{e.Priority, e.WantType}
+
+	m, ok := l.cids[e.Cid]
+	if !ok {
+		m = make(map[peer.ID]entry)
+		l.cids[e.Cid] = m
+	}
+	m[p] = entry{e.Priority, e.WantType}
+}
+
+// CancelWant returns true if the cid was present in the wantlist.
+func (l *peerLedger) CancelWant(p peer.ID, k cid.Cid) bool {
+	wants, ok := l.peers[p]
+	if !ok {
+		return false
+	}
+	delete(wants, k)
+	if len(wants) == 0 {
+		delete(l.peers, p)
+	}
+
+	l.removePeerFromCid(p, k)
+	return true
+}
+
+// CancelWantWithType will not cancel WantBlock if we sent a HAVE message.
+func (l *peerLedger) CancelWantWithType(p peer.ID, k cid.Cid, typ pb.Message_Wantlist_WantType) {
+	wants, ok := l.peers[p]
 	if !ok {
-		m = make(map[peer.ID]struct{})
-		l.cids[k] = m
+		return
+	}
+	e, ok := wants[k]
+	if !ok {
+		return
 	}
-	m[p] = struct{}{}
+	if typ == pb.Message_Wantlist_Have && e.WantType == pb.Message_Wantlist_Block {
+		return
+	}
+
+	delete(wants, k)
+	if len(wants) == 0 {
+		delete(l.peers, p)
+	}
+
+	l.removePeerFromCid(p, k)
 }
 
-func (l *peerLedger) CancelWant(p peer.ID, k cid.Cid) {
+func (l *peerLedger) removePeerFromCid(p peer.ID, k cid.Cid) {
 	m, ok := l.cids[k]
 	if !ok {
 		return
 	}
 	delete(m, p)
 	if len(m) == 0 {
 		delete(l.cids, k)
 	}
 }
 
-func (l *peerLedger) Peers(k cid.Cid) []peer.ID {
+type entryForPeer struct {
+	Peer peer.ID
+	entry
+}
+
+type entry struct {
+	Priority int32
+	WantType pb.Message_Wantlist_WantType
+}
+
+func (l *peerLedger) Peers(k cid.Cid) []entryForPeer {
 	m, ok := l.cids[k]
 	if !ok {
 		return nil
 	}
-	peers := make([]peer.ID, 0, len(m))
-	for p := range m {
+	peers := make([]entryForPeer, 0, len(m))
+	for p, e := range m {
+		peers = append(peers, entryForPeer{p, e})
+	}
+	return peers
+}
+
+func (l *peerLedger) CollectPeerIDs() []peer.ID {
+	peers := make([]peer.ID, 0, len(l.peers))
+	for p := range l.peers {
 		peers = append(peers, p)
 	}
 	return peers
 }
+
+func (l *peerLedger) WantlistSizeForPeer(p peer.ID) int {
+	return len(l.peers[p])
+}
+
+func (l *peerLedger) WantlistForPeer(p peer.ID) []wl.Entry {
+	cids, ok := l.peers[p]
+	if !ok {
+		return nil
+	}
+
+	entries := make([]wl.Entry, 0, len(l.cids))
+	for c, e := range cids {
+		entries = append(entries, wl.Entry{
+			Cid:      c,
+			Priority: e.Priority,
+			WantType: e.WantType,
+		})
+	}
+	return entries
+}
+
+// ClearPeerWantlist does not take an effort to fully erase it from memory.
+// This is intended when the peer is still connected and the map capacity could
+// be reused. If the memory should be freed use PeerDisconnected instead.
+func (l *peerLedger) ClearPeerWantlist(p peer.ID) {
+	cids, ok := l.peers[p]
+	if !ok {
+		return
+	}
+
+	for c := range cids {
+		l.removePeerFromCid(p, c)
+	}
+}
+
+func (l *peerLedger) PeerDisconnected(p peer.ID) {
+	l.ClearPeerWantlist(p)
+	delete(l.peers, p)
+}
diff --git a/bitswap/server/server.go b/bitswap/server/server.go
index 136ae3d..4244560 100644
--- a/bitswap/server/server.go
+++ b/bitswap/server/server.go
@@ -208,10 +208,20 @@ func MaxOutstandingBytesPerPeer(count int) Option {
 	return func(bs *Server) {
 		bs.engineOptions = append(bs.engineOptions, o)
 	}
 }
 
+// MaxQueuedWantlistEntriesPerPeer limits how much individual entries each peer is allowed to send.
+// If a peer send us more than this we will truncate newest entries.
+// It defaults to defaults.MaxQueuedWantlistEntiresPerPeer.
+func MaxQueuedWantlistEntriesPerPeer(count uint) Option {
+	o := decision.WithMaxQueuedWantlistEntriesPerPeer(count)
+	return func(bs *Server) {
+		bs.engineOptions = append(bs.engineOptions, o)
+	}
+}
+
 // HasBlockBufferSize configure how big the new blocks buffer should be.
 func HasBlockBufferSize(count int) Option {
 	if count < 0 {
 		panic("cannot have negative buffer size")
 	}
diff --git a/examples/go.sum b/examples/go.sum
index a336c0e..49755f7 100644
--- a/examples/go.sum
+++ b/examples/go.sum
@@ -418,12 +418,12 @@ github.com/ipfs/go-ipfs-files v0.0.3/go.mod h1:INEFm0LL2LWXBhNJ2PMIIb2w45hpXgPjN
 github.com/ipfs/go-ipfs-files v0.3.0 h1:fallckyc5PYjuMEitPNrjRfpwl7YFt69heCOUhsbGxQ=
 github.com/ipfs/go-ipfs-files v0.3.0/go.mod h1:xAUtYMwB+iu/dtf6+muHNSFQCJG2dSiStR2P6sn9tIM=
 github.com/ipfs/go-ipfs-posinfo v0.0.1 h1:Esoxj+1JgSjX0+ylc0hUmJCOv6V2vFoZiETLR6OtpRs=
 github.com/ipfs/go-ipfs-posinfo v0.0.1/go.mod h1:SwyeVP+jCwiDu0C313l/8jg6ZxM0qqtlt2a0vILTc1A=
 github.com/ipfs/go-ipfs-pq v0.0.1/go.mod h1:LWIqQpqfRG3fNc5XsnIhz/wQ2XXGyugQwls7BgUmUfY=
-github.com/ipfs/go-ipfs-pq v0.0.2 h1:e1vOOW6MuOwG2lqxcLA+wEn93i/9laCY8sXAw76jFOY=
 github.com/ipfs/go-ipfs-pq v0.0.2/go.mod h1:LWIqQpqfRG3fNc5XsnIhz/wQ2XXGyugQwls7BgUmUfY=
+github.com/ipfs/go-ipfs-pq v0.0.3 h1:YpoHVJB+jzK15mr/xsWC574tyDLkezVrDNeaalQBsTE=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1 h1:Io++k0Vf/wK+tfnhEh63Yte1oQK5VGT2hIEYpD0Rzx8=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1/go.mod h1:tAwRjCV0RjLTjH8DR/AU7VYvfQECg+lpUy2Mdzv7gyk=
 github.com/ipfs/go-ipfs-routing v0.1.0/go.mod h1:hYoUkJLyAUKhF58tysKpids8RNDPO42BVMgK5dNsoqY=
 github.com/ipfs/go-ipfs-routing v0.2.1/go.mod h1:xiNNiwgjmLqPS1cimvAw6EyB9rkVDbiocA4yY+wRNLM=
 github.com/ipfs/go-ipfs-routing v0.3.0 h1:9W/W3N+g+y4ZDeffSgqhgo7BsBSJwPMcyssET9OWevc=
@@ -471,11 +471,11 @@ github.com/ipfs/go-namesys v0.7.0 h1:xqosk71GIVRkFDtF2UNRcXn4LdNeo7tzuy8feHD6NbU
 github.com/ipfs/go-namesys v0.7.0/go.mod h1:KYSZBVZG3VJC34EfqqJPG7T48aWgxseoMPAPA5gLyyQ=
 github.com/ipfs/go-path v0.3.0 h1:tkjga3MtpXyM5v+3EbRvOHEoo+frwi4oumw5K+KYWyA=
 github.com/ipfs/go-path v0.3.0/go.mod h1:NOScsVgxfC/eIw4nz6OiGwK42PjaSJ4Y/ZFPn1Xe07I=
 github.com/ipfs/go-peertaskqueue v0.1.0/go.mod h1:Jmk3IyCcfl1W3jTW3YpghSwSEC6IJ3Vzz/jUmWw8Z0U=
 github.com/ipfs/go-peertaskqueue v0.7.0/go.mod h1:M/akTIE/z1jGNXMU7kFB4TeSEFvj68ow0Rrb04donIU=
-github.com/ipfs/go-peertaskqueue v0.8.0 h1:JyNO144tfu9bx6Hpo119zvbEL9iQ760FHOiJYsUjqaU=
+github.com/ipfs/go-peertaskqueue v0.8.1 h1:YhxAs1+wxb5jk7RvS0LHdyiILpNmRIRnZVztekOF0pg=
 github.com/ipfs/go-unixfs v0.2.4/go.mod h1:SUdisfUjNoSDzzhGVxvCL9QO/nKdwXdr+gbMUdqcbYw=
 github.com/ipfs/go-unixfs v0.3.1 h1:LrfED0OGfG98ZEegO4/xiprx2O+yS+krCMQSp7zLVv8=
 github.com/ipfs/go-unixfs v0.3.1/go.mod h1:h4qfQYzghiIc8ZNFKiLMFWOTzrWIAtzYQ59W/pCFf1o=
 github.com/ipfs/go-unixfsnode v1.1.2/go.mod h1:5dcE2x03pyjHk4JjamXmunTMzz+VUtqvPwZjIEkfV6s=
 github.com/ipfs/go-unixfsnode v1.5.1 h1:JcR3t5C2nM1V7PMzhJ/Qmo19NkoFIKweDSZyDx+CjkI=
diff --git a/go.mod b/go.mod
index ade83a3..53d1eec 100644
--- a/go.mod
+++ b/go.mod
@@ -31,11 +31,11 @@ require (
 	github.com/ipfs/go-log/v2 v2.5.1
 	github.com/ipfs/go-merkledag v0.9.0
 	github.com/ipfs/go-metrics-interface v0.0.1
 	github.com/ipfs/go-namesys v0.7.0
 	github.com/ipfs/go-path v0.3.0
-	github.com/ipfs/go-peertaskqueue v0.8.0
+	github.com/ipfs/go-peertaskqueue v0.8.1
 	github.com/ipfs/go-unixfs v0.3.1
 	github.com/ipfs/go-unixfsnode v1.5.1
 	github.com/ipfs/interface-go-ipfs-core v0.10.0
 	github.com/ipld/go-car v0.5.0
 	github.com/ipld/go-car/v2 v2.5.1
@@ -81,11 +81,11 @@ require (
 	github.com/ipfs/bbloom v0.0.4 // indirect
 	github.com/ipfs/go-bitfield v1.0.0 // indirect
 	github.com/ipfs/go-block-format v0.1.1 // indirect
 	github.com/ipfs/go-ipfs-ds-help v1.1.0 // indirect
 	github.com/ipfs/go-ipfs-files v0.3.0 // indirect
-	github.com/ipfs/go-ipfs-pq v0.0.2 // indirect
+	github.com/ipfs/go-ipfs-pq v0.0.3 // indirect
 	github.com/ipfs/go-ipld-cbor v0.0.6 // indirect
 	github.com/ipfs/go-ipld-legacy v0.1.1 // indirect
 	github.com/ipfs/go-verifcid v0.0.2 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.3 // indirect
diff --git a/go.sum b/go.sum
index 9def56b..da272b6 100644
--- a/go.sum
+++ b/go.sum
@@ -423,12 +423,13 @@ github.com/ipfs/go-ipfs-files v0.0.3/go.mod h1:INEFm0LL2LWXBhNJ2PMIIb2w45hpXgPjN
 github.com/ipfs/go-ipfs-files v0.3.0 h1:fallckyc5PYjuMEitPNrjRfpwl7YFt69heCOUhsbGxQ=
 github.com/ipfs/go-ipfs-files v0.3.0/go.mod h1:xAUtYMwB+iu/dtf6+muHNSFQCJG2dSiStR2P6sn9tIM=
 github.com/ipfs/go-ipfs-posinfo v0.0.1 h1:Esoxj+1JgSjX0+ylc0hUmJCOv6V2vFoZiETLR6OtpRs=
 github.com/ipfs/go-ipfs-posinfo v0.0.1/go.mod h1:SwyeVP+jCwiDu0C313l/8jg6ZxM0qqtlt2a0vILTc1A=
 github.com/ipfs/go-ipfs-pq v0.0.1/go.mod h1:LWIqQpqfRG3fNc5XsnIhz/wQ2XXGyugQwls7BgUmUfY=
-github.com/ipfs/go-ipfs-pq v0.0.2 h1:e1vOOW6MuOwG2lqxcLA+wEn93i/9laCY8sXAw76jFOY=
 github.com/ipfs/go-ipfs-pq v0.0.2/go.mod h1:LWIqQpqfRG3fNc5XsnIhz/wQ2XXGyugQwls7BgUmUfY=
+github.com/ipfs/go-ipfs-pq v0.0.3 h1:YpoHVJB+jzK15mr/xsWC574tyDLkezVrDNeaalQBsTE=
+github.com/ipfs/go-ipfs-pq v0.0.3/go.mod h1:btNw5hsHBpRcSSgZtiNm/SLj5gYIZ18AKtv3kERkRb4=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1 h1:Io++k0Vf/wK+tfnhEh63Yte1oQK5VGT2hIEYpD0Rzx8=
 github.com/ipfs/go-ipfs-redirects-file v0.1.1/go.mod h1:tAwRjCV0RjLTjH8DR/AU7VYvfQECg+lpUy2Mdzv7gyk=
 github.com/ipfs/go-ipfs-routing v0.1.0/go.mod h1:hYoUkJLyAUKhF58tysKpids8RNDPO42BVMgK5dNsoqY=
 github.com/ipfs/go-ipfs-routing v0.2.1/go.mod h1:xiNNiwgjmLqPS1cimvAw6EyB9rkVDbiocA4yY+wRNLM=
 github.com/ipfs/go-ipfs-routing v0.3.0 h1:9W/W3N+g+y4ZDeffSgqhgo7BsBSJwPMcyssET9OWevc=
@@ -477,12 +478,12 @@ github.com/ipfs/go-namesys v0.7.0 h1:xqosk71GIVRkFDtF2UNRcXn4LdNeo7tzuy8feHD6NbU
 github.com/ipfs/go-namesys v0.7.0/go.mod h1:KYSZBVZG3VJC34EfqqJPG7T48aWgxseoMPAPA5gLyyQ=
 github.com/ipfs/go-path v0.3.0 h1:tkjga3MtpXyM5v+3EbRvOHEoo+frwi4oumw5K+KYWyA=
 github.com/ipfs/go-path v0.3.0/go.mod h1:NOScsVgxfC/eIw4nz6OiGwK42PjaSJ4Y/ZFPn1Xe07I=
 github.com/ipfs/go-peertaskqueue v0.1.0/go.mod h1:Jmk3IyCcfl1W3jTW3YpghSwSEC6IJ3Vzz/jUmWw8Z0U=
 github.com/ipfs/go-peertaskqueue v0.7.0/go.mod h1:M/akTIE/z1jGNXMU7kFB4TeSEFvj68ow0Rrb04donIU=
-github.com/ipfs/go-peertaskqueue v0.8.0 h1:JyNO144tfu9bx6Hpo119zvbEL9iQ760FHOiJYsUjqaU=
-github.com/ipfs/go-peertaskqueue v0.8.0/go.mod h1:cz8hEnnARq4Du5TGqiWKgMr/BOSQ5XOgMOh1K5YYKKM=
+github.com/ipfs/go-peertaskqueue v0.8.1 h1:YhxAs1+wxb5jk7RvS0LHdyiILpNmRIRnZVztekOF0pg=
+github.com/ipfs/go-peertaskqueue v0.8.1/go.mod h1:Oxxd3eaK279FxeydSPPVGHzbwVeHjatZ2GA8XD+KbPU=
 github.com/ipfs/go-unixfs v0.2.4/go.mod h1:SUdisfUjNoSDzzhGVxvCL9QO/nKdwXdr+gbMUdqcbYw=
 github.com/ipfs/go-unixfs v0.3.1 h1:LrfED0OGfG98ZEegO4/xiprx2O+yS+krCMQSp7zLVv8=
 github.com/ipfs/go-unixfs v0.3.1/go.mod h1:h4qfQYzghiIc8ZNFKiLMFWOTzrWIAtzYQ59W/pCFf1o=
 github.com/ipfs/go-unixfsnode v1.1.2/go.mod h1:5dcE2x03pyjHk4JjamXmunTMzz+VUtqvPwZjIEkfV6s=
 github.com/ipfs/go-unixfsnode v1.5.1 h1:JcR3t5C2nM1V7PMzhJ/Qmo19NkoFIKweDSZyDx+CjkI=

Vulnerable code: 	// ledgerMap lists block-related Ledgers by their Partner key.
	ledgerMap map[peer.ID]*ledger

		ledgerMap:                       make(map[peer.ID]*ledger),
	partner := e.findOrCreate(p)

	partner.lk.Lock()
	entries := partner.wantList.Entries()
	partner.lk.Unlock()
	return entries
	response := make([]peer.ID, 0, len(e.ledgerMap))

	for _, ledger := range e.ledgerMap {
		response = append(response, ledger.Partner)
	}
	return response
	for _, entry := range wants {
		e.peerLedger.Wants(p, entry.Cid)
	}
	for _, entry := range cancels {
		e.peerLedger.CancelWant(p, entry.Cid)
	}
	e.lock.Unlock()
	// Get the ledger for the peer
	l := e.findOrCreate(p)
	l.lk.Lock()
	defer l.lk.Unlock()

	// If the peer sent a full wantlist, replace the ledger's wantlist
		l.wantList = wl.New()
	var activeEntries []peertask.Task
	// Remove cancelled blocks from the queue
		if l.CancelWant(entry.Cid) {
		// Add each want-have / want-block to the ledger
		l.Wants(c, entry.Priority, entry.WantType)

		e.peerRequestQueue.PushTasks(p, activeEntries...)
	l := e.findOrCreate(from)

	l.lk.Lock()
	defer l.lk.Unlock()
		e.scoreLedger.AddToReceivedBytes(l.Partner, len(blk.RawData()))
	missingWants := make(map[peer.ID][]cid.Cid)
		for _, p := range peers {
			e.lock.RLock()
			ledger, ok := e.ledgerMap[p]
			e.lock.RUnlock()

			if !ok {
				// This can happen if the peer has disconnected while we're processing this list.
				log.Debugw("failed to find peer in ledger", "peer", p)
				missingWants[p] = append(missingWants[p], k)
				continue
			}
			ledger.lk.RLock()
			entry, ok := ledger.WantListContains(k)
			ledger.lk.RUnlock()
			if !ok {
				// This can happen if the peer has canceled their want while we're processing this message.
				log.Debugw("wantlist index doesn't match peer's wantlist", "peer", p)
				missingWants[p] = append(missingWants[p], k)
				continue
			}
			e.peerRequestQueue.PushTasks(p, peertask.Task{
				Topic:    entry.Cid,
	// If we found missing wants (e.g., because the peer disconnected, we have some races here)
	// remove them from the list. Unfortunately, we still have to re-check because the user
	// could have re-connected in the meantime.
	if len(missingWants) > 0 {
		e.lock.Lock()
		for p, wl := range missingWants {
			if ledger, ok := e.ledgerMap[p]; ok {
				ledger.lk.RLock()
				for _, k := range wl {
					if _, has := ledger.WantListContains(k); has {
						continue
					}
					e.peerLedger.CancelWant(p, k)
				}
				ledger.lk.RUnlock()
			} else {
				for _, k := range wl {
					e.peerLedger.CancelWant(p, k)
				}
			}
		}
		e.lock.Unlock()
	}

	l := e.findOrCreate(p)
	l.lk.Lock()
	defer l.lk.Unlock()
		e.scoreLedger.AddToSentBytes(l.Partner, len(block.RawData()))
		l.wantList.RemoveType(block.Cid(), pb.Message_Wantlist_Block)
			l.wantList.RemoveType(bp.Cid, pb.Message_Wantlist_Have)
	_, ok := e.ledgerMap[p]
	if !ok {
		e.ledgerMap[p] = newLedger(p)
	}

	ledger, ok := e.ledgerMap[p]
	if ok {
		ledger.lk.RLock()
		entries := ledger.Entries()
		ledger.lk.RUnlock()

		for _, entry := range entries {
			e.peerLedger.CancelWant(p, entry.Cid)
		}
	}
	delete(e.ledgerMap, p)

// ledger lazily instantiates a ledger
func (e *Engine) findOrCreate(p peer.ID) *ledger {
	// Take a read lock (as it's less expensive) to check if we have a ledger
	// for the peer
	e.lock.RLock()
	l, ok := e.ledgerMap[p]
	e.lock.RUnlock()
	if ok {
		return l
	}

	// There's no ledger, so take a write lock, then check again and create the
	// ledger if necessary
	e.lock.Lock()
	defer e.lock.Unlock()
	l, ok = e.ledgerMap[p]
	if !ok {
		l = newLedger(p)
		e.ledgerMap[p] = l
	}
	return l
}

func TestPeerIsAddedToPeersWhenMessageReceivedOrSent(t *testing.T) {
	sanfrancisco.Engine.MessageSent(seattle.Peer, m)
	if !peerIsPartner(seattle.Peer, sanfrancisco.Engine) {
		t.Fatal("Peer wasn't added as a Partner")
	}

	if entries[0].Priority != 4 || entries[1].Priority != 3 || entries[2].Priority != 2 || entries[3].Priority != 1 {
		t.Fatal("expected wantlist to be sorted")
	}

	cids map[cid.Cid]map[peer.ID]struct{}
	return &peerLedger{cids: make(map[cid.Cid]map[peer.ID]struct{})}
func (l *peerLedger) Wants(p peer.ID, k cid.Cid) {
	m, ok := l.cids[k]
		m = make(map[peer.ID]struct{})
		l.cids[k] = m
	m[p] = struct{}{}
func (l *peerLedger) CancelWant(p peer.ID, k cid.Cid) {
func (l *peerLedger) Peers(k cid.Cid) []peer.ID {
	peers := make([]peer.ID, 0, len(m))
	for p := range m {
github.com/ipfs/go-ipfs-pq v0.0.2 h1:e1vOOW6MuOwG2lqxcLA+wEn93i/9laCY8sXAw76jFOY=
github.com/ipfs/go-peertaskqueue v0.8.0 h1:JyNO144tfu9bx6Hpo119zvbEL9iQ760FHOiJYsUjqaU=
	github.com/ipfs/go-peertaskqueue v0.8.0
	github.com/ipfs/go-ipfs-pq v0.0.2 // indirect
github.com/ipfs/go-ipfs-pq v0.0.2 h1:e1vOOW6MuOwG2lqxcLA+wEn93i/9laCY8sXAw76jFOY=
github.com/ipfs/go-peertaskqueue v0.8.0 h1:JyNO144tfu9bx6Hpo119zvbEL9iQ760FHOiJYsUjqaU=
github.com/ipfs/go-peertaskqueue v0.8.0/go.mod h1:cz8hEnnARq4Du5TGqiWKgMr/BOSQ5XOgMOh1K5YYKKM=
Fixed code: 
	// Maximum size of the wantlist we are willing to keep in memory.
	MaxQueuedWantlistEntiresPerPeer = 1024
func MaxQueuedWantlistEntriesPerPeer(count uint) Option {
	return Option{server.MaxQueuedWantlistEntriesPerPeer(count)}
}

	"math/bits"

	maxQueuedWantlistEntriesPerPeer uint
// WithMaxQueuedWantlistEntriesPerPeer limits how much individual entries each peer is allowed to send.
// If a peer send us more than this we will truncate newest entries.
// It defaults to DefaultMaxQueuedWantlistEntiresPerPeer.
func WithMaxQueuedWantlistEntriesPerPeer(count uint) Option {
	return func(e *Engine) {
		e.maxQueuedWantlistEntriesPerPeer = count
	}
}

		maxQueuedWantlistEntriesPerPeer: defaults.MaxQueuedWantlistEntiresPerPeer,
	e.lock.RLock()
	defer e.lock.RUnlock()
	return e.peerLedger.WantlistForPeer(p)
	return e.peerLedger.CollectPeerIDs()
		e.peerLedger.ClearPeerWantlist(p)
	s := uint(e.peerLedger.WantlistSizeForPeer(p))
	if wouldBe := s + uint(len(wants)); wouldBe > e.maxQueuedWantlistEntriesPerPeer {
		log.Debugw("wantlist overflow", "local", e.self, "remote", p, "would be", wouldBe)
		// truncate wantlist to avoid overflow
		available, o := bits.Sub(e.maxQueuedWantlistEntriesPerPeer, s, 0)
		if o != 0 {
			available = 0
		}
		wants = wants[:available]
	}
	for _, entry := range wants {
		e.peerLedger.Wants(p, entry.Entry)
	}
		if e.peerLedger.CancelWant(p, entry.Cid) {
	e.lock.Unlock()

	var activeEntries []peertask.Task
		e.peerRequestQueue.PushTasksTruncated(e.maxQueuedWantlistEntriesPerPeer, p, activeEntries...)
		e.scoreLedger.AddToReceivedBytes(from, len(blk.RawData()))
		for _, entry := range peers {
			e.peerRequestQueue.PushTasksTruncated(e.maxQueuedWantlistEntriesPerPeer, entry.Peer, peertask.Task{
				Topic:    k,
	e.lock.Lock()
	defer e.lock.Unlock()
		e.scoreLedger.AddToSentBytes(p, len(block.RawData()))
		e.peerLedger.CancelWantWithType(p, block.Cid(), pb.Message_Wantlist_Block)
			e.peerLedger.CancelWantWithType(p, bp.Cid, pb.Message_Wantlist_Have)
	e.peerRequestQueue.Clear(p)

	e.peerLedger.PeerDisconnected(p)
func TestPeerIsAddedToPeersWhenMessageSent(t *testing.T) {
	// We need to request something for it to add us as partner.
	m.AddEntry(blocks.NewBlock([]byte("Hæ")).Cid(), 0, pb.Message_Wantlist_Block, true)

	wl "github.com/ipfs/go-libipfs/bitswap/client/wantlist"
	pb "github.com/ipfs/go-libipfs/bitswap/message/pb"

	// thoses two maps are inversions of each other
	peers map[peer.ID]map[cid.Cid]entry
	cids  map[cid.Cid]map[peer.ID]entry
	return &peerLedger{
		peers: make(map[peer.ID]map[cid.Cid]entry),
		cids:  make(map[cid.Cid]map[peer.ID]entry),
	}
func (l *peerLedger) Wants(p peer.ID, e wl.Entry) {
	cids, ok := l.peers[p]
	if !ok {
		cids = make(map[cid.Cid]entry)
		l.peers[p] = cids
	}
	cids[e.Cid] = entry{e.Priority, e.WantType}

	m, ok := l.cids[e.Cid]
	if !ok {
		m = make(map[peer.ID]entry)
		l.cids[e.Cid] = m
	}
	m[p] = entry{e.Priority, e.WantType}
}

// CancelWant returns true if the cid was present in the wantlist.
func (l *peerLedger) CancelWant(p peer.ID, k cid.Cid) bool {
	wants, ok := l.peers[p]
	if !ok {
		return false
	}
	delete(wants, k)
	if len(wants) == 0 {
		delete(l.peers, p)
	}

	l.removePeerFromCid(p, k)
	return true
}

// CancelWantWithType will not cancel WantBlock if we sent a HAVE message.
func (l *peerLedger) CancelWantWithType(p peer.ID, k cid.Cid, typ pb.Message_Wantlist_WantType) {
	wants, ok := l.peers[p]
		return
	}
	e, ok := wants[k]
	if !ok {
		return
	if typ == pb.Message_Wantlist_Have && e.WantType == pb.Message_Wantlist_Block {
		return
	}

	delete(wants, k)
	if len(wants) == 0 {
		delete(l.peers, p)
	}

	l.removePeerFromCid(p, k)
func (l *peerLedger) removePeerFromCid(p peer.ID, k cid.Cid) {
type entryForPeer struct {
	Peer peer.ID
	entry
}

type entry struct {
	Priority int32
	WantType pb.Message_Wantlist_WantType
}

func (l *peerLedger) Peers(k cid.Cid) []entryForPeer {
	peers := make([]entryForPeer, 0, len(m))
	for p, e := range m {
		peers = append(peers, entryForPeer{p, e})
	}
	return peers
}

func (l *peerLedger) CollectPeerIDs() []peer.ID {
	peers := make([]peer.ID, 0, len(l.peers))
	for p := range l.peers {

func (l *peerLedger) WantlistSizeForPeer(p peer.ID) int {
	return len(l.peers[p])
}

func (l *peerLedger) WantlistForPeer(p peer.ID) []wl.Entry {
	cids, ok := l.peers[p]
	if !ok {
		return nil
	}

	entries := make([]wl.Entry, 0, len(l.cids))
	for c, e := range cids {
		entries = append(entries, wl.Entry{
			Cid:      c,
			Priority: e.Priority,
			WantType: e.WantType,
		})
	}
	return entries
}

// ClearPeerWantlist does not take an effort to fully erase it from memory.
// This is intended when the peer is still connected and the map capacity could
// be reused. If the memory should be freed use PeerDisconnected instead.
func (l *peerLedger) ClearPeerWantlist(p peer.ID) {
	cids, ok := l.peers[p]
	if !ok {
		return
	}

	for c := range cids {
		l.removePeerFromCid(p, c)
	}
}

func (l *peerLedger) PeerDisconnected(p peer.ID) {
	l.ClearPeerWantlist(p)
	delete(l.peers, p)
}
// MaxQueuedWantlistEntriesPerPeer limits how much individual entries each peer is allowed to send.
// If a peer send us more than this we will truncate newest entries.
// It defaults to defaults.MaxQueuedWantlistEntiresPerPeer.
func MaxQueuedWantlistEntriesPerPeer(count uint) Option {
	o := decision.WithMaxQueuedWantlistEntriesPerPeer(count)
	return func(bs *Server) {
		bs.engineOptions = append(bs.engineOptions, o)
	}
}

github.com/ipfs/go-ipfs-pq v0.0.3 h1:YpoHVJB+jzK15mr/xsWC574tyDLkezVrDNeaalQBsTE=
github.com/ipfs/go-peertaskqueue v0.8.1 h1:YhxAs1+wxb5jk7RvS0LHdyiILpNmRIRnZVztekOF0pg=
	github.com/ipfs/go-peertaskqueue v0.8.1
	github.com/ipfs/go-ipfs-pq v0.0.3 // indirect
github.com/ipfs/go-ipfs-pq v0.0.3 h1:YpoHVJB+jzK15mr/xsWC574tyDLkezVrDNeaalQBsTE=
github.com/ipfs/go-ipfs-pq v0.0.3/go.mod h1:btNw5hsHBpRcSSgZtiNm/SLj5gYIZ18AKtv3kERkRb4=
github.com/ipfs/go-peertaskqueue v0.8.1 h1:YhxAs1+wxb5jk7RvS0LHdyiILpNmRIRnZVztekOF0pg=
github.com/ipfs/go-peertaskqueue v0.8.1/go.mod h1:Oxxd3eaK279FxeydSPPVGHzbwVeHjatZ2GA8XD+KbPU=