Skip to content

Commit 91a6ceb

Browse files
hannahhowardhannahhoward
committed
feat(graphsync): add link limits
add options to configure the maximum number of allowed links to traverse on the requestor and the responder
1 parent 57df640 commit 91a6ceb

13 files changed

+182
-25
lines changed

‎go.mod

+1-1
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ require (
2727
github.com/ipfs/go-peertaskqueue v0.2.0
2828
github.com/ipfs/go-unixfs v0.2.4
2929
github.com/ipld/go-codec-dagpb v1.3.0
30-
github.com/ipld/go-ipld-prime v0.12.0
30+
github.com/ipld/go-ipld-prime v0.12.3-0.20210929125341-05d5528bd84e
3131
github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c
3232
github.com/libp2p/go-buffer-pool v0.0.2
3333
github.com/libp2p/go-libp2p v0.13.0

‎go.sum

+4-2
Original file line numberDiff line numberDiff line change
@@ -238,8 +238,8 @@ github.com/ipfs/go-verifcid v0.0.1/go.mod h1:5Hrva5KBeIog4A+UpqlaIU+DEstipcJYQQZ
238238
github.com/ipld/go-codec-dagpb v1.3.0 h1:czTcaoAuNNyIYWs6Qe01DJ+sEX7B+1Z0LcXjSatMGe8=
239239
github.com/ipld/go-codec-dagpb v1.3.0/go.mod h1:ga4JTU3abYApDC3pZ00BC2RSvC3qfBb9MSJkMLSwnhA=
240240
github.com/ipld/go-ipld-prime v0.11.0/go.mod h1:+WIAkokurHmZ/KwzDOMUuoeJgaRQktHtEaLglS3ZeV8=
241-
github.com/ipld/go-ipld-prime v0.12.0 h1:JapyKWTsJgmhrPI7hfx4V798c/RClr85sXfBZnH1VIw=
242-
github.com/ipld/go-ipld-prime v0.12.0/go.mod h1:hy8b93WleDMRKumOJnTIrr0MbbFbx9GD6Kzxa53Xppc=
241+
github.com/ipld/go-ipld-prime v0.12.3-0.20210929125341-05d5528bd84e h1:HPLQ9V/OFHKjfbFio8vQV+EW7lpQPj+iPl93VcwSTYM=
242+
github.com/ipld/go-ipld-prime v0.12.3-0.20210929125341-05d5528bd84e/go.mod h1:PaeLYq8k6dJLmDUSLrzkEpoGV4PEfe/1OtFN/eALOc8=
243243
github.com/jackpal/gateway v1.0.5/go.mod h1:lTpwd4ACLXmpyiCTRtfiNyVnUmqT9RivzCDQetPfnjA=
244244
github.com/jackpal/go-nat-pmp v1.0.1/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
245245
github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus=
@@ -539,6 +539,8 @@ github.com/multiformats/go-multiaddr-net v0.2.0/go.mod h1:gGdH3UXny6U3cKKYCvpXI5
539539
github.com/multiformats/go-multibase v0.0.1/go.mod h1:bja2MqRZ3ggyXtZSEDKpl0uO/gviWFaSteVbWT51qgs=
540540
github.com/multiformats/go-multibase v0.0.3 h1:l/B6bJDQjvQ5G52jw4QGSYeOTZoAwIO77RblWplfIqk=
541541
github.com/multiformats/go-multibase v0.0.3/go.mod h1:5+1R4eQrT3PkYZ24C3W2Ue2tPwIdYQD509ZjSb5y9Oc=
542+
github.com/multiformats/go-multicodec v0.3.0 h1:tstDwfIjiHbnIjeM5Lp+pMrSeN+LCMsEwOrkPmWm03A=
543+
github.com/multiformats/go-multicodec v0.3.0/go.mod h1:qGGaQmioCDh+TeFOnxrbU0DaIPw8yFgAZgFG0V7p1qQ=
542544
github.com/multiformats/go-multihash v0.0.1/go.mod h1:w/5tugSrLEbWqlcgJabL3oHFKTwfvkofsjW2Qa1ct4U=
543545
github.com/multiformats/go-multihash v0.0.5/go.mod h1:lt/HCbqlQwlPBz7lv0sQCdtfcMtlJvakRUn/0Ual8po=
544546
github.com/multiformats/go-multihash v0.0.8/go.mod h1:YSLudS+Pi8NHE7o6tb3D8vrpKa63epEDmG8nTduyAew=

‎impl/graphsync.go

+20-2
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,8 @@ type graphsyncConfigOptions struct {
7474
maxInProgressIncomingRequests uint64
7575
maxInProgressOutgoingRequests uint64
7676
registerDefaultValidator bool
77+
maxLinksPerOutgoingRequest uint64
78+
maxLinksPerIncomingRequest uint64
7779
}
7880

7981
// Option defines the functional option type that can be used to configure
@@ -136,6 +138,22 @@ func MaxInProgressOutgoingRequests(maxInProgressOutgoingRequests uint64) Option
136138
}
137139
}
138140

141+
// MaxLinksPerOutgoingRequests changes the allowed number of links an outgoing
142+
// request can traverse before failing
143+
func MaxLinksPerOutgoingRequests(maxLinksPerOutgoingRequest uint64) Option {
144+
return func(gs *graphsyncConfigOptions) {
145+
gs.maxLinksPerOutgoingRequest = maxLinksPerOutgoingRequest
146+
}
147+
}
148+
149+
// MaxLinksPerIncomingRequests changes the allowed number of links an incoming
150+
// request can traverse before failing
151+
func MaxLinksPerIncomingRequests(maxLinksPerIncomingRequest uint64) Option {
152+
return func(gs *graphsyncConfigOptions) {
153+
gs.maxLinksPerIncomingRequest = maxLinksPerIncomingRequest
154+
}
155+
}
156+
139157
// New creates a new GraphSync Exchange on the given network,
140158
// and the given link loader+storer.
141159
func New(parent context.Context, network gsnet.GraphSyncNetwork,
@@ -179,11 +197,11 @@ func New(parent context.Context, network gsnet.GraphSyncNetwork,
179197

180198
asyncLoader := asyncloader.New(ctx, linkSystem, requestAllocator)
181199
requestQueue := taskqueue.NewTaskQueue(ctx)
182-
requestManager := requestmanager.New(ctx, asyncLoader, linkSystem, outgoingRequestHooks, incomingResponseHooks, networkErrorListeners, requestQueue, network.ConnectionManager())
200+
requestManager := requestmanager.New(ctx, asyncLoader, linkSystem, outgoingRequestHooks, incomingResponseHooks, networkErrorListeners, requestQueue, network.ConnectionManager(), gsConfig.maxLinksPerOutgoingRequest)
183201
requestExecutor := executor.NewExecutor(requestManager, incomingBlockHooks, asyncLoader.AsyncLoad)
184202
responseAssembler := responseassembler.New(ctx, peerManager)
185203
peerTaskQueue := peertaskqueue.New()
186-
responseManager := responsemanager.New(ctx, linkSystem, responseAssembler, peerTaskQueue, requestQueuedHooks, incomingRequestHooks, outgoingBlockHooks, requestUpdatedHooks, completedResponseListeners, requestorCancelledListeners, blockSentListeners, networkErrorListeners, gsConfig.maxInProgressIncomingRequests, network.ConnectionManager())
204+
responseManager := responsemanager.New(ctx, linkSystem, responseAssembler, peerTaskQueue, requestQueuedHooks, incomingRequestHooks, outgoingBlockHooks, requestUpdatedHooks, completedResponseListeners, requestorCancelledListeners, blockSentListeners, networkErrorListeners, gsConfig.maxInProgressIncomingRequests, network.ConnectionManager(), gsConfig.maxLinksPerIncomingRequest)
187205
graphSync := &GraphSync{
188206
network: network,
189207
linkSystem: linkSystem,

‎impl/graphsync_test.go

+52
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,58 @@ func TestRejectRequestsByDefault(t *testing.T) {
173173
testutil.VerifySingleTerminalError(ctx, t, errChan)
174174
}
175175

176+
func TestGraphsyncRoundTripRequestBudgetRequestor(t *testing.T) {
177+
// create network
178+
ctx := context.Background()
179+
ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
180+
defer cancel()
181+
td := newGsTestData(ctx, t)
182+
183+
var linksToTraverse uint64 = 5
184+
// initialize graphsync on first node to make requests
185+
requestor := td.GraphSyncHost1(MaxLinksPerOutgoingRequests(linksToTraverse))
186+
187+
// setup receiving peer to just record message coming in
188+
blockChainLength := 100
189+
blockChain := testutil.SetupBlockChain(ctx, t, td.persistence2, 100, blockChainLength)
190+
191+
// initialize graphsync on second node to response to requests
192+
td.GraphSyncHost2()
193+
194+
progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)
195+
196+
// response budgets don't include the root block, so total links traverse with be one more than expected
197+
blockChain.VerifyResponseRange(ctx, progressChan, 0, int(linksToTraverse))
198+
testutil.VerifySingleTerminalError(ctx, t, errChan)
199+
require.Len(t, td.blockStore1, int(linksToTraverse), "did not store all blocks")
200+
}
201+
202+
func TestGraphsyncRoundTripRequestBudgetResponder(t *testing.T) {
203+
// create network
204+
ctx := context.Background()
205+
ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
206+
defer cancel()
207+
td := newGsTestData(ctx, t)
208+
209+
var linksToTraverse uint64 = 5
210+
// initialize graphsync on first node to make requests
211+
requestor := td.GraphSyncHost1()
212+
213+
// setup receiving peer to just record message coming in
214+
blockChainLength := 100
215+
blockChain := testutil.SetupBlockChain(ctx, t, td.persistence2, 100, blockChainLength)
216+
217+
// initialize graphsync on second node to response to requests
218+
td.GraphSyncHost2(MaxLinksPerIncomingRequests(linksToTraverse))
219+
220+
progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)
221+
222+
// response budgets don't include the root block, so total links traverse with be one more than expected
223+
blockChain.VerifyResponseRange(ctx, progressChan, 0, int(linksToTraverse))
224+
testutil.VerifySingleTerminalError(ctx, t, errChan)
225+
require.Len(t, td.blockStore1, int(linksToTraverse), "did not store all blocks")
226+
}
227+
176228
func TestGraphsyncRoundTrip(t *testing.T) {
177229
// create network
178230
ctx := context.Background()

‎ipldutil/traverser.go

+10
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ type TraversalBuilder struct {
3636
Visitor traversal.AdvVisitFn
3737
LinkSystem ipld.LinkSystem
3838
Chooser traversal.LinkTargetNodePrototypeChooser
39+
Budget *traversal.Budget
3940
}
4041

4142
// Traverser is an interface for performing a selector traversal that operates iteratively --
@@ -81,6 +82,7 @@ func (tb TraversalBuilder) Start(parentCtx context.Context) Traverser {
8182
visitor: defaultVisitor,
8283
chooser: defaultChooser,
8384
linkSystem: tb.LinkSystem,
85+
budget: tb.Budget,
8486
awaitRequest: make(chan struct{}, 1),
8587
stateChan: make(chan state, 1),
8688
responses: make(chan nextResponse),
@@ -120,6 +122,7 @@ type traverser struct {
120122
chooser traversal.LinkTargetNodePrototypeChooser
121123
currentLink ipld.Link
122124
currentContext ipld.LinkContext
125+
budget *traversal.Budget
123126
isDone bool
124127
completionErr error
125128
awaitRequest chan struct{}
@@ -184,6 +187,12 @@ func (t *traverser) start() {
184187
t.writeDone(err)
185188
return
186189
}
190+
if t.budget != nil {
191+
t.budget.LinkBudget--
192+
if t.budget.LinkBudget <= 0 {
193+
t.writeDone(&traversal.ErrBudgetExceeded{BudgetKind: "link", Link: t.root})
194+
}
195+
}
187196
nd, err := t.linkSystem.Load(ipld.LinkContext{Ctx: t.ctx}, t.root, ns)
188197
if err != nil {
189198
t.writeDone(err)
@@ -201,6 +210,7 @@ func (t *traverser) start() {
201210
LinkSystem: t.linkSystem,
202211
LinkTargetNodePrototypeChooser: t.chooser,
203212
},
213+
Budget: t.budget,
204214
}.WalkAdv(nd, sel, t.visitor)
205215
t.writeDone(err)
206216
}()

‎ipldutil/traverser_test.go

+54-4
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package ipldutil
33
import (
44
"bytes"
55
"context"
6+
"math"
67
"testing"
78
"time"
89

@@ -56,7 +57,7 @@ func TestTraverser(t *testing.T) {
5657
testdata.MiddleMapBlock,
5758
testdata.LeafAlphaBlock,
5859
testdata.LeafAlphaBlock,
59-
})
60+
}, nil)
6061
})
6162

6263
t.Run("traverses correctly, blockchain", func(t *testing.T) {
@@ -86,13 +87,58 @@ func TestTraverser(t *testing.T) {
8687
blockChain.VerifyWholeChainWithTypes(ctx, inProgressChan)
8788
close(done)
8889
}()
89-
checkTraverseSequence(ctx, t, traverser, blockChain.AllBlocks())
90+
checkTraverseSequence(ctx, t, traverser, blockChain.AllBlocks(), nil)
9091
close(inProgressChan)
9192
testutil.AssertDoesReceive(ctx, t, done, "should have completed verification but did not")
9293
})
94+
95+
t.Run("errors correctly, with budget", func(t *testing.T) {
96+
store := make(map[ipld.Link][]byte)
97+
persistence := testutil.NewTestStore(store)
98+
blockChain := testutil.SetupBlockChain(ctx, t, persistence, 100, 10)
99+
traverser := TraversalBuilder{
100+
Root: blockChain.TipLink,
101+
Selector: blockChain.Selector(),
102+
Chooser: blockChain.Chooser,
103+
LinkSystem: persistence,
104+
Visitor: func(tp traversal.Progress, node ipld.Node, r traversal.VisitReason) error {
105+
return nil
106+
},
107+
Budget: &traversal.Budget{
108+
NodeBudget: math.MaxInt64,
109+
LinkBudget: 6,
110+
},
111+
}.Start(ctx)
112+
var path ipld.Path
113+
for i := 0; i < 6; i++ {
114+
path = path.AppendSegment(ipld.PathSegmentOfString("Parents"))
115+
path = path.AppendSegment(ipld.PathSegmentOfInt(0))
116+
}
117+
checkTraverseSequence(ctx, t, traverser, blockChain.Blocks(0, 6), &traversal.ErrBudgetExceeded{BudgetKind: "link", Path: path, Link: blockChain.LinkTipIndex(6)})
118+
})
119+
120+
t.Run("errors correctly, no budget", func(t *testing.T) {
121+
store := make(map[ipld.Link][]byte)
122+
persistence := testutil.NewTestStore(store)
123+
blockChain := testutil.SetupBlockChain(ctx, t, persistence, 100, 10)
124+
traverser := TraversalBuilder{
125+
Root: blockChain.TipLink,
126+
Selector: blockChain.Selector(),
127+
Chooser: blockChain.Chooser,
128+
LinkSystem: persistence,
129+
Visitor: func(tp traversal.Progress, node ipld.Node, r traversal.VisitReason) error {
130+
return nil
131+
},
132+
Budget: &traversal.Budget{
133+
NodeBudget: math.MaxInt64,
134+
LinkBudget: 0,
135+
},
136+
}.Start(ctx)
137+
checkTraverseSequence(ctx, t, traverser, []blocks.Block{}, &traversal.ErrBudgetExceeded{BudgetKind: "link", Link: blockChain.TipLink})
138+
})
93139
}
94140

95-
func checkTraverseSequence(ctx context.Context, t *testing.T, traverser Traverser, expectedBlks []blocks.Block) {
141+
func checkTraverseSequence(ctx context.Context, t *testing.T, traverser Traverser, expectedBlks []blocks.Block, finalErr error) {
96142
for _, blk := range expectedBlks {
97143
isComplete, err := traverser.IsComplete()
98144
require.False(t, isComplete)
@@ -104,5 +150,9 @@ func checkTraverseSequence(ctx context.Context, t *testing.T, traverser Traverse
104150
}
105151
isComplete, err := traverser.IsComplete()
106152
require.True(t, isComplete)
107-
require.NoError(t, err)
153+
if finalErr == nil {
154+
require.NoError(t, err)
155+
} else {
156+
require.EqualError(t, err, finalErr.Error())
157+
}
108158
}

‎requestmanager/client.go

+12-9
Original file line numberDiff line numberDiff line change
@@ -87,15 +87,16 @@ type AsyncLoader interface {
8787
// RequestManager tracks outgoing requests and processes incoming reponses
8888
// to them.
8989
type RequestManager struct {
90-
ctx context.Context
91-
cancel func()
92-
messages chan requestManagerMessage
93-
peerHandler PeerHandler
94-
rc *responseCollector
95-
asyncLoader AsyncLoader
96-
disconnectNotif *pubsub.PubSub
97-
linkSystem ipld.LinkSystem
98-
connManager network.ConnManager
90+
ctx context.Context
91+
cancel func()
92+
messages chan requestManagerMessage
93+
peerHandler PeerHandler
94+
rc *responseCollector
95+
asyncLoader AsyncLoader
96+
disconnectNotif *pubsub.PubSub
97+
linkSystem ipld.LinkSystem
98+
connManager network.ConnManager
99+
maxLinksPerRequest uint64
99100

100101
// dont touch out side of run loop
101102
nextRequestID graphsync.RequestID
@@ -129,6 +130,7 @@ func New(ctx context.Context,
129130
networkErrorListeners *listeners.NetworkErrorListeners,
130131
requestQueue taskqueue.TaskQueue,
131132
connManager network.ConnManager,
133+
maxLinksPerRequest uint64,
132134
) *RequestManager {
133135
ctx, cancel := context.WithCancel(ctx)
134136
return &RequestManager{
@@ -145,6 +147,7 @@ func New(ctx context.Context,
145147
networkErrorListeners: networkErrorListeners,
146148
requestQueue: requestQueue,
147149
connManager: connManager,
150+
maxLinksPerRequest: maxLinksPerRequest,
148151
}
149152
}
150153

‎requestmanager/requestmanager_test.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -1019,7 +1019,7 @@ func newTestData(ctx context.Context, t *testing.T) *testData {
10191019
td.networkErrorListeners = listeners.NewNetworkErrorListeners()
10201020
td.taskqueue = taskqueue.NewTaskQueue(ctx)
10211021
lsys := cidlink.DefaultLinkSystem()
1022-
td.requestManager = New(ctx, td.fal, lsys, td.requestHooks, td.responseHooks, td.networkErrorListeners, td.taskqueue, td.tcm)
1022+
td.requestManager = New(ctx, td.fal, lsys, td.requestHooks, td.responseHooks, td.networkErrorListeners, td.taskqueue, td.tcm, 0)
10231023
td.executor = executor.NewExecutor(td.requestManager, td.blockHooks, td.fal.AsyncLoad)
10241024
td.requestManager.SetDelegate(td.fph)
10251025
td.requestManager.Startup()

‎requestmanager/server.go

+8
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,13 @@ func (rm *RequestManager) requestTask(requestID graphsync.RequestID) executor.Re
102102
var initialRequest bool
103103
if ipr.traverser == nil {
104104
initialRequest = true
105+
var budget *traversal.Budget
106+
if rm.maxLinksPerRequest > 0 {
107+
budget = &traversal.Budget{
108+
NodeBudget: math.MaxInt64,
109+
LinkBudget: int64(rm.maxLinksPerRequest),
110+
}
111+
}
105112
ipr.traverser = ipldutil.TraversalBuilder{
106113
Root: cidlink.Link{Cid: ipr.request.Root()},
107114
Selector: ipr.request.Selector(),
@@ -118,6 +125,7 @@ func (rm *RequestManager) requestTask(requestID graphsync.RequestID) executor.Re
118125
},
119126
Chooser: ipr.nodeStyleChooser,
120127
LinkSystem: rm.linkSystem,
128+
Budget: budget,
121129
}.Start(ipr.ctx)
122130
}
123131

‎responsemanager/client.go

+3
Original file line numberDiff line numberDiff line change
@@ -155,6 +155,7 @@ type ResponseManager struct {
155155
inProgressResponses map[responseKey]*inProgressResponseStatus
156156
maxInProcessRequests uint64
157157
connManager network.ConnManager
158+
maxLinksPerRequest uint64
158159
}
159160

160161
// New creates a new response manager for responding to requests
@@ -172,6 +173,7 @@ func New(ctx context.Context,
172173
networkErrorListeners NetworkErrorListeners,
173174
maxInProcessRequests uint64,
174175
connManager network.ConnManager,
176+
maxLinksPerRequest uint64,
175177
) *ResponseManager {
176178
ctx, cancelFn := context.WithCancel(ctx)
177179
messages := make(chan responseManagerMessage, 16)
@@ -194,6 +196,7 @@ func New(ctx context.Context,
194196
inProgressResponses: make(map[responseKey]*inProgressResponseStatus),
195197
maxInProcessRequests: maxInProcessRequests,
196198
connManager: connManager,
199+
maxLinksPerRequest: maxLinksPerRequest,
197200
}
198201
rm.qe = &queryExecutor{
199202
blockHooks: blockHooks,

0 commit comments

Comments
 (0)