graphsync_test.go 31.4 KB
Newer Older
1 2 3
package graphsync

import (
4
	"bytes"
5
	"context"
6
	"errors"
7
	"fmt"
8 9
	"io"
	"io/ioutil"
10 11
	"math"
	"math/rand"
12 13
	"os"
	"path/filepath"
14 15 16
	"testing"
	"time"

Hannah Howard's avatar
Hannah Howard committed
17
	basicnode "github.com/ipld/go-ipld-prime/node/basic"
Hannah Howard's avatar
Hannah Howard committed
18
	"github.com/stretchr/testify/require"
19

20
	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
21

22
	blocks "github.com/ipfs/go-block-format"
23
	"github.com/ipfs/go-blockservice"
24
	"github.com/ipfs/go-cid"
25 26 27 28 29 30 31 32 33 34 35 36
	"github.com/ipfs/go-datastore"
	dss "github.com/ipfs/go-datastore/sync"
	bstore "github.com/ipfs/go-ipfs-blockstore"
	chunker "github.com/ipfs/go-ipfs-chunker"
	offline "github.com/ipfs/go-ipfs-exchange-offline"
	files "github.com/ipfs/go-ipfs-files"
	ipldformat "github.com/ipfs/go-ipld-format"
	"github.com/ipfs/go-merkledag"
	unixfile "github.com/ipfs/go-unixfs/file"
	"github.com/ipfs/go-unixfs/importer/balanced"
	ihelper "github.com/ipfs/go-unixfs/importer/helpers"

37
	"github.com/ipfs/go-graphsync"
38

39
	"github.com/ipfs/go-graphsync/cidset"
40
	"github.com/ipfs/go-graphsync/ipldutil"
41 42 43 44
	gsmsg "github.com/ipfs/go-graphsync/message"
	gsnet "github.com/ipfs/go-graphsync/network"
	"github.com/ipfs/go-graphsync/testutil"
	ipld "github.com/ipld/go-ipld-prime"
45
	"github.com/ipld/go-ipld-prime/traversal/selector"
Edgar Lee's avatar
Edgar Lee committed
46
	ipldselector "github.com/ipld/go-ipld-prime/traversal/selector"
47
	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
48
	"github.com/libp2p/go-libp2p-core/host"
49
	"github.com/libp2p/go-libp2p-core/peer"
50 51 52 53 54 55 56 57
	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
)

func TestMakeRequestToNetwork(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
58
	td := newGsTestData(ctx, t)
59
	r := &receiver{
60
		messageReceived: make(chan receivedMessage),
61
	}
62 63
	td.gsnet2.SetDelegate(r)
	graphSync := td.GraphSyncHost1()
64

65
	blockChainLength := 100
66
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader1, td.storer1, 100, blockChainLength)
67

68 69
	requestCtx, requestCancel := context.WithCancel(ctx)
	defer requestCancel()
70
	graphSync.Request(requestCtx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)
71

72
	var message receivedMessage
Hannah Howard's avatar
Hannah Howard committed
73
	testutil.AssertReceive(ctx, t, r.messageReceived, &message, "did not receive message sent")
74

75
	sender := message.sender
Hannah Howard's avatar
Hannah Howard committed
76
	require.Equal(t, td.host1.ID(), sender, "received message from wrong node")
77

78
	received := message.message
79
	receivedRequests := received.Requests()
Hannah Howard's avatar
Hannah Howard committed
80
	require.Len(t, receivedRequests, 1, "Did not add request to received message")
81
	receivedRequest := receivedRequests[0]
82
	receivedSpec := receivedRequest.Selector()
Hannah Howard's avatar
Hannah Howard committed
83
	require.Equal(t, blockChain.Selector(), receivedSpec, "did not transmit selector spec correctly")
84
	_, err := ipldutil.ParseSelector(receivedSpec)
Hannah Howard's avatar
Hannah Howard committed
85
	require.NoError(t, err, "did not receive parsible selector on other side")
86

87
	returnedData, found := receivedRequest.Extension(td.extensionName)
Hannah Howard's avatar
Hannah Howard committed
88 89
	require.True(t, found)
	require.Equal(t, td.extensionData, returnedData, "Failed to encode extension")
90
}
91 92 93 94 95 96

func TestSendResponseToIncomingRequest(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()
97
	td := newGsTestData(ctx, t)
98 99 100
	r := &receiver{
		messageReceived: make(chan receivedMessage),
	}
101
	td.gsnet1.SetDelegate(r)
102 103

	var receivedRequestData []byte
104
	// initialize graphsync on second node to response to requests
105
	gsnet := td.GraphSyncHost2()
106 107
	gsnet.RegisterIncomingRequestHook(
		func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
108
			var has bool
109
			receivedRequestData, has = requestData.Extension(td.extensionName)
Hannah Howard's avatar
Hannah Howard committed
110
			require.True(t, has, "did not have expected extension")
111
			hookActions.SendExtensionData(td.extensionResponse)
112 113
		},
	)
114

115
	blockChainLength := 100
116
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, 100, blockChainLength)
117

118
	requestID := graphsync.RequestID(rand.Int31())
119 120

	message := gsmsg.New()
121
	message.AddRequest(gsmsg.NewRequest(requestID, blockChain.TipLink.(cidlink.Link).Cid, blockChain.Selector(), graphsync.Priority(math.MaxInt32), td.extension))
122
	// send request across network
123
	err := td.gsnet1.SendMessage(ctx, td.host2.ID(), message)
Hannah Howard's avatar
Hannah Howard committed
124
	require.NoError(t, err)
125 126 127
	// read the values sent back to requestor
	var received gsmsg.GraphSyncMessage
	var receivedBlocks []blocks.Block
128
	var receivedExtensions [][]byte
129
	for {
Hannah Howard's avatar
Hannah Howard committed
130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146
		var message receivedMessage
		testutil.AssertReceive(ctx, t, r.messageReceived, &message, "did not receive complete response")

		sender := message.sender
		require.Equal(t, td.host2.ID(), sender, "received message from wrong node")

		received = message.message
		receivedBlocks = append(receivedBlocks, received.Blocks()...)
		receivedResponses := received.Responses()
		receivedExtension, found := receivedResponses[0].Extension(td.extensionName)
		if found {
			receivedExtensions = append(receivedExtensions, receivedExtension)
		}
		require.Len(t, receivedResponses, 1, "Did not receive response")
		require.Equal(t, requestID, receivedResponses[0].RequestID(), "Sent response for incorrect request id")
		if receivedResponses[0].Status() != graphsync.PartialResponse {
			break
147 148 149
		}
	}

Hannah Howard's avatar
Hannah Howard committed
150 151 152 153
	require.Len(t, receivedBlocks, blockChainLength, "Send incorrect number of blocks or there were duplicate blocks")
	require.Equal(t, td.extensionData, receivedRequestData, "did not receive correct request extension data")
	require.Len(t, receivedExtensions, 1, "should have sent extension responses but didn't")
	require.Equal(t, td.extensionResponseData, receivedExtensions[0], "did not return correct extension data")
154
}
155

156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176
func TestRejectRequestsByDefault(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	requestor := td.GraphSyncHost1()
	// setup responder to disable default validation, meaning all requests are rejected
	_ = td.GraphSyncHost2(RejectAllRequestsByDefault())

	blockChainLength := 5
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, 5, blockChainLength)

	// send request across network
	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)

	testutil.VerifyEmptyResponse(ctx, t, progressChan)
	testutil.VerifySingleTerminalError(ctx, t, errChan)
}

177 178 179 180 181
func TestGraphsyncRoundTrip(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()
182
	td := newGsTestData(ctx, t)
183

184
	// initialize graphsync on first node to make requests
185
	requestor := td.GraphSyncHost1()
186 187

	// setup receiving peer to just record message coming in
188
	blockChainLength := 100
189
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, 100, blockChainLength)
190 191

	// initialize graphsync on second node to response to requests
192
	responder := td.GraphSyncHost2()
193

194
	var receivedResponseData []byte
195 196
	var receivedRequestData []byte

197
	requestor.RegisterIncomingResponseHook(
Hannah Howard's avatar
Hannah Howard committed
198
		func(p peer.ID, responseData graphsync.ResponseData, hookActions graphsync.IncomingResponseHookActions) {
199
			data, has := responseData.Extension(td.extensionName)
200
			if has {
201
				receivedResponseData = data
202 203 204
			}
		})

205
	responder.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
206
		var has bool
207
		receivedRequestData, has = requestData.Extension(td.extensionName)
208 209 210
		if !has {
			hookActions.TerminateWithError(errors.New("Missing extension"))
		} else {
211
			hookActions.SendExtensionData(td.extensionResponse)
212 213
		}
	})
214

215
	finalResponseStatusChan := make(chan graphsync.ResponseStatusCode, 1)
216
	responder.RegisterCompletedResponseListener(func(p peer.ID, request graphsync.RequestData, status graphsync.ResponseStatusCode) {
217 218 219 220 221
		select {
		case finalResponseStatusChan <- status:
		default:
		}
	})
222
	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)
223

Hannah Howard's avatar
Hannah Howard committed
224 225 226
	blockChain.VerifyWholeChain(ctx, progressChan)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
	require.Len(t, td.blockStore1, blockChainLength, "did not store all blocks")
227 228

	// verify extension roundtrip
Hannah Howard's avatar
Hannah Howard committed
229
	require.Equal(t, td.extensionData, receivedRequestData, "did not receive correct extension request data")
230
	require.Equal(t, td.extensionResponseData, receivedResponseData, "did not receive correct extension response data")
231

232
	// verify listener
233 234 235
	var finalResponseStatus graphsync.ResponseStatusCode
	testutil.AssertReceive(ctx, t, finalResponseStatusChan, &finalResponseStatus, "should receive status")
	require.Equal(t, graphsync.RequestCompletedFull, finalResponseStatus)
236
}
237

238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258
func TestGraphsyncRoundTripPartial(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()

	// setup an IPLD tree and put all but 1 node into the second nodes block store
	tree := testutil.NewTestIPLDTree()
	td.blockStore2[tree.LeafAlphaLnk] = tree.LeafAlphaBlock.RawData()
	td.blockStore2[tree.MiddleMapNodeLnk] = tree.MiddleMapBlock.RawData()
	td.blockStore2[tree.MiddleListNodeLnk] = tree.MiddleListBlock.RawData()
	td.blockStore2[tree.RootNodeLnk] = tree.RootBlock.RawData()

	// initialize graphsync on second node to response to requests
	responder := td.GraphSyncHost2()

	finalResponseStatusChan := make(chan graphsync.ResponseStatusCode, 1)
259
	responder.RegisterCompletedResponseListener(func(p peer.ID, request graphsync.RequestData, status graphsync.ResponseStatusCode) {
260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280
		select {
		case finalResponseStatusChan <- status:
		default:
		}
	})
	// create a selector to traverse the whole tree
	ssb := builder.NewSelectorSpecBuilder(basicnode.Style.Any)
	allSelector := ssb.ExploreRecursive(selector.RecursionLimitDepth(10),
		ssb.ExploreAll(ssb.ExploreRecursiveEdge())).Node()

	_, errChan := requestor.Request(ctx, td.host2.ID(), tree.RootNodeLnk, allSelector)

	for err := range errChan {
		// verify the error is received for leaf beta node being missing
		require.EqualError(t, err, fmt.Sprintf("Remote Peer Is Missing Block: %s", tree.LeafBetaLnk.String()))
	}
	require.Equal(t, tree.LeafAlphaBlock.RawData(), td.blockStore1[tree.LeafAlphaLnk])
	require.Equal(t, tree.MiddleListBlock.RawData(), td.blockStore1[tree.MiddleListNodeLnk])
	require.Equal(t, tree.MiddleMapBlock.RawData(), td.blockStore1[tree.MiddleMapNodeLnk])
	require.Equal(t, tree.RootBlock.RawData(), td.blockStore1[tree.RootNodeLnk])

281
	// verify listener
282 283 284 285 286
	var finalResponseStatus graphsync.ResponseStatusCode
	testutil.AssertReceive(ctx, t, finalResponseStatusChan, &finalResponseStatus, "should receive status")
	require.Equal(t, graphsync.RequestCompletedPartial, finalResponseStatus)
}

287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335
func TestGraphsyncRoundTripIgnoreCids(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()

	// setup receiving peer to just record message coming in
	blockChainLength := 100
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, 100, blockChainLength)

	firstHalf := blockChain.Blocks(0, 50)
	set := cid.NewSet()
	for _, blk := range firstHalf {
		td.blockStore1[cidlink.Link{Cid: blk.Cid()}] = blk.RawData()
		set.Add(blk.Cid())
	}
	encodedCidSet, err := cidset.EncodeCidSet(set)
	require.NoError(t, err)
	extension := graphsync.ExtensionData{
		Name: graphsync.ExtensionDoNotSendCIDs,
		Data: encodedCidSet,
	}

	// initialize graphsync on second node to response to requests
	responder := td.GraphSyncHost2()

	totalSent := 0
	totalSentOnWire := 0
	responder.RegisterOutgoingBlockHook(func(p peer.ID, requestData graphsync.RequestData, blockData graphsync.BlockData, hookActions graphsync.OutgoingBlockHookActions) {
		totalSent++
		if blockData.BlockSizeOnWire() > 0 {
			totalSentOnWire++
		}
	})

	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), extension)

	blockChain.VerifyWholeChain(ctx, progressChan)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
	require.Len(t, td.blockStore1, blockChainLength, "did not store all blocks")

	require.Equal(t, blockChainLength, totalSent)
	require.Equal(t, blockChainLength-set.Len(), totalSentOnWire)
}

336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386
func TestPauseResume(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()

	// setup receiving peer to just record message coming in
	blockChainLength := 100
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, 100, blockChainLength)

	// initialize graphsync on second node to response to requests
	responder := td.GraphSyncHost2()

	stopPoint := 50
	blocksSent := 0
	requestIDChan := make(chan graphsync.RequestID, 1)
	responder.RegisterOutgoingBlockHook(func(p peer.ID, requestData graphsync.RequestData, blockData graphsync.BlockData, hookActions graphsync.OutgoingBlockHookActions) {
		_, has := requestData.Extension(td.extensionName)
		if has {
			select {
			case requestIDChan <- requestData.ID():
			default:
			}
			blocksSent++
			if blocksSent == stopPoint {
				hookActions.PauseResponse()
			}
		} else {
			hookActions.TerminateWithError(errors.New("should have sent extension"))
		}
	})

	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)

	blockChain.VerifyResponseRange(ctx, progressChan, 0, stopPoint)
	timer := time.NewTimer(100 * time.Millisecond)
	testutil.AssertDoesReceiveFirst(t, timer.C, "should pause request", progressChan)

	requestID := <-requestIDChan
	err := responder.UnpauseResponse(td.host1.ID(), requestID)
	require.NoError(t, err)

	blockChain.VerifyRemainder(ctx, progressChan, stopPoint)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
	require.Len(t, td.blockStore1, blockChainLength, "did not store all blocks")

}
387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432
func TestPauseResumeRequest(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()

	// setup receiving peer to just record message coming in
	blockChainLength := 100
	blockSize := 100
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, uint64(blockSize), blockChainLength)

	// initialize graphsync on second node to response to requests
	_ = td.GraphSyncHost2()

	stopPoint := 50
	blocksReceived := 0
	requestIDChan := make(chan graphsync.RequestID, 1)
	requestor.RegisterIncomingBlockHook(func(p peer.ID, responseData graphsync.ResponseData, blockData graphsync.BlockData, hookActions graphsync.IncomingBlockHookActions) {
		select {
		case requestIDChan <- responseData.RequestID():
		default:
		}
		blocksReceived++
		if blocksReceived == stopPoint {
			hookActions.PauseRequest()
		}
	})

	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)

	blockChain.VerifyResponseRange(ctx, progressChan, 0, stopPoint-1)
	timer := time.NewTimer(100 * time.Millisecond)
	testutil.AssertDoesReceiveFirst(t, timer.C, "should pause request", progressChan)

	requestID := <-requestIDChan
	err := requestor.UnpauseRequest(requestID, td.extensionUpdate)
	require.NoError(t, err)

	blockChain.VerifyRemainder(ctx, progressChan, stopPoint-1)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
	require.Len(t, td.blockStore1, blockChainLength, "did not store all blocks")
}
433

Hannah Howard's avatar
Hannah Howard committed
434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476
func TestPauseResumeViaUpdate(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	var receivedReponseData []byte
	var receivedUpdateData []byte
	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()

	requestor.RegisterIncomingResponseHook(func(p peer.ID, response graphsync.ResponseData, hookActions graphsync.IncomingResponseHookActions) {
		if response.Status() == graphsync.RequestPaused {
			var has bool
			receivedReponseData, has = response.Extension(td.extensionName)
			if has {
				hookActions.UpdateRequestWithExtensions(td.extensionUpdate)
			}
		}
	})

	// setup receiving peer to just record message coming in
	blockChainLength := 100
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, 100, blockChainLength)

	// initialize graphsync on second node to response to requests
	responder := td.GraphSyncHost2()
	stopPoint := 50
	blocksSent := 0
	responder.RegisterOutgoingBlockHook(func(p peer.ID, requestData graphsync.RequestData, blockData graphsync.BlockData, hookActions graphsync.OutgoingBlockHookActions) {
		_, has := requestData.Extension(td.extensionName)
		if has {
			blocksSent++
			if blocksSent == stopPoint {
				hookActions.SendExtensionData(td.extensionResponse)
				hookActions.PauseResponse()
			}
		} else {
			hookActions.TerminateWithError(errors.New("should have sent extension"))
		}
	})
	responder.RegisterRequestUpdatedHook(func(p peer.ID, request graphsync.RequestData, update graphsync.RequestData, hookActions graphsync.RequestUpdatedHookActions) {
Hannah Howard's avatar
Hannah Howard committed
477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537
		var has bool
		receivedUpdateData, has = update.Extension(td.extensionName)
		if has {
			hookActions.UnpauseResponse()
		}
	})
	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)

	blockChain.VerifyWholeChain(ctx, progressChan)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
	require.Len(t, td.blockStore1, blockChainLength, "did not store all blocks")

	require.Equal(t, td.extensionResponseData, receivedReponseData, "did not receive correct extension response data")
	require.Equal(t, td.extensionUpdateData, receivedUpdateData, "did not receive correct extension update data")
}

func TestPauseResumeViaUpdateOnBlockHook(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	var receivedReponseData []byte
	var receivedUpdateData []byte
	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()

	// setup receiving peer to just record message coming in
	blockChainLength := 100
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader2, td.storer2, 100, blockChainLength)

	stopPoint := 50
	blocksReceived := 0
	requestor.RegisterIncomingBlockHook(func(p peer.ID, response graphsync.ResponseData, block graphsync.BlockData, hookActions graphsync.IncomingBlockHookActions) {
		blocksReceived++
		if response.Status() == graphsync.RequestPaused && blocksReceived == stopPoint {
			var has bool
			receivedReponseData, has = response.Extension(td.extensionName)
			if has {
				hookActions.UpdateRequestWithExtensions(td.extensionUpdate)
			}
		}
	})

	// initialize graphsync on second node to response to requests
	responder := td.GraphSyncHost2()
	blocksSent := 0
	responder.RegisterOutgoingBlockHook(func(p peer.ID, requestData graphsync.RequestData, blockData graphsync.BlockData, hookActions graphsync.OutgoingBlockHookActions) {
		_, has := requestData.Extension(td.extensionName)
		if has {
			blocksSent++
			if blocksSent == stopPoint {
				hookActions.SendExtensionData(td.extensionResponse)
				hookActions.PauseResponse()
			}
		} else {
			hookActions.TerminateWithError(errors.New("should have sent extension"))
		}
	})
	responder.RegisterRequestUpdatedHook(func(p peer.ID, request graphsync.RequestData, update graphsync.RequestData, hookActions graphsync.RequestUpdatedHookActions) {
Hannah Howard's avatar
Hannah Howard committed
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553
		var has bool
		receivedUpdateData, has = update.Extension(td.extensionName)
		if has {
			hookActions.UnpauseResponse()
		}
	})
	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), td.extension)

	blockChain.VerifyWholeChain(ctx, progressChan)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
	require.Len(t, td.blockStore1, blockChainLength, "did not store all blocks")

	require.Equal(t, td.extensionResponseData, receivedReponseData, "did not receive correct extension response data")
	require.Equal(t, td.extensionUpdateData, receivedUpdateData, "did not receive correct extension update data")
}

554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592
func TestGraphsyncRoundTripAlternatePersistenceAndNodes(t *testing.T) {
	// create network
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
	defer cancel()
	td := newGsTestData(ctx, t)

	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()

	// initialize graphsync on second node to response to requests
	responder := td.GraphSyncHost2()

	// alternate storing location for responder
	altStore1 := make(map[ipld.Link][]byte)
	altLoader1, altStorer1 := testutil.NewTestStore(altStore1)

	// alternate storing location for requestor
	altStore2 := make(map[ipld.Link][]byte)
	altLoader2, altStorer2 := testutil.NewTestStore(altStore2)

	err := requestor.RegisterPersistenceOption("chainstore", altLoader1, altStorer1)
	require.NoError(t, err)

	err = responder.RegisterPersistenceOption("chainstore", altLoader2, altStorer2)
	require.NoError(t, err)

	blockChainLength := 100
	blockChain := testutil.SetupBlockChain(ctx, t, altLoader1, altStorer2, 100, blockChainLength)

	extensionName := graphsync.ExtensionName("blockchain")
	extension := graphsync.ExtensionData{
		Name: extensionName,
		Data: nil,
	}

	requestor.RegisterOutgoingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.OutgoingRequestHookActions) {
		_, has := requestData.Extension(extensionName)
		if has {
Hannah Howard's avatar
Hannah Howard committed
593
			hookActions.UseLinkTargetNodeStyleChooser(blockChain.Chooser)
594 595 596 597 598 599
			hookActions.UsePersistenceOption("chainstore")
		}
	})
	responder.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
		_, has := requestData.Extension(extensionName)
		if has {
Hannah Howard's avatar
Hannah Howard committed
600
			hookActions.UseLinkTargetNodeStyleChooser(blockChain.Chooser)
601 602 603 604 605 606
			hookActions.UsePersistenceOption("chainstore")
		}
	})

	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector())
	testutil.VerifyEmptyResponse(ctx, t, progressChan)
607
	testutil.VerifyHasErrors(ctx, t, errChan)
608 609 610 611 612 613 614 615 616

	progressChan, errChan = requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector(), extension)

	blockChain.VerifyWholeChainWithTypes(ctx, progressChan)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
	require.Len(t, td.blockStore1, 0, "should store no blocks in normal store")
	require.Len(t, altStore1, blockChainLength, "did not store all blocks in alternate store")
}

617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632
// TestRoundTripLargeBlocksSlowNetwork test verifies graphsync continues to work
// under a specific of adverse conditions:
// -- large blocks being returned by a query
// -- slow network connection
// It verifies that Graphsync will properly break up network message packets
// so they can still be decoded on the client side, instead of building up a huge
// backlog of blocks and then sending them in one giant network packet that can't
// be decoded on the client side
func TestRoundTripLargeBlocksSlowNetwork(t *testing.T) {
	// create network
	if testing.Short() {
		t.Skip()
	}
	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
	defer cancel()
633 634 635 636 637
	td := newGsTestData(ctx, t)
	td.mn.SetLinkDefaults(mocknet.LinkOptions{Latency: 100 * time.Millisecond, Bandwidth: 3000000})

	// initialize graphsync on first node to make requests
	requestor := td.GraphSyncHost1()
638

639 640
	// setup receiving peer to just record message coming in
	blockChainLength := 40
641
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader1, td.storer2, 200000, blockChainLength)
642 643 644 645

	// initialize graphsync on second node to response to requests
	td.GraphSyncHost2()

646
	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector())
647

Hannah Howard's avatar
Hannah Howard committed
648 649
	blockChain.VerifyWholeChain(ctx, progressChan)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
650 651
}

652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713
// What this test does:
// - Construct a blockstore + dag service
// - Import a file to UnixFS v1
// - setup a graphsync request from one node to the other
// for the file
// - Load the file from the new block store on the other node
// using the
// existing UnixFS v1 file reader
// - Verify the bytes match the original
func TestUnixFSFetch(t *testing.T) {
	if testing.Short() {
		t.Skip()
	}

	const unixfsChunkSize uint64 = 1 << 10
	const unixfsLinksPerLevel = 1024

	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
	defer cancel()

	makeLoader := func(bs bstore.Blockstore) ipld.Loader {
		return func(lnk ipld.Link, lnkCtx ipld.LinkContext) (io.Reader, error) {
			c, ok := lnk.(cidlink.Link)
			if !ok {
				return nil, errors.New("Incorrect Link Type")
			}
			// read block from one store
			block, err := bs.Get(c.Cid)
			if err != nil {
				return nil, err
			}
			return bytes.NewReader(block.RawData()), nil
		}
	}

	makeStorer := func(bs bstore.Blockstore) ipld.Storer {
		return func(lnkCtx ipld.LinkContext) (io.Writer, ipld.StoreCommitter, error) {
			var buf bytes.Buffer
			var committer ipld.StoreCommitter = func(lnk ipld.Link) error {
				c, ok := lnk.(cidlink.Link)
				if !ok {
					return errors.New("Incorrect Link Type")
				}
				block, err := blocks.NewBlockWithCid(buf.Bytes(), c.Cid)
				if err != nil {
					return err
				}
				return bs.Put(block)
			}
			return &buf, committer, nil
		}
	}
	// make a blockstore and dag service
	bs1 := bstore.NewBlockstore(dss.MutexWrap(datastore.NewMapDatastore()))

	// make a second blockstore
	bs2 := bstore.NewBlockstore(dss.MutexWrap(datastore.NewMapDatastore()))
	dagService2 := merkledag.NewDAGService(blockservice.New(bs2, offline.Exchange(bs2)))

	// read in a fixture file
	path, err := filepath.Abs(filepath.Join("fixtures", "lorem.txt"))
Hannah Howard's avatar
Hannah Howard committed
714
	require.NoError(t, err, "unable to create path for fixture file")
715 716

	f, err := os.Open(path)
Hannah Howard's avatar
Hannah Howard committed
717 718
	require.NoError(t, err, "unable to open fixture file")

719 720 721 722 723 724 725 726 727 728 729 730 731 732 733
	var buf bytes.Buffer
	tr := io.TeeReader(f, &buf)
	file := files.NewReaderFile(tr)

	// import to UnixFS
	bufferedDS := ipldformat.NewBufferedDAG(ctx, dagService2)

	params := ihelper.DagBuilderParams{
		Maxlinks:   unixfsLinksPerLevel,
		RawLeaves:  true,
		CidBuilder: nil,
		Dagserv:    bufferedDS,
	}

	db, err := params.New(chunker.NewSizeSplitter(file, int64(unixfsChunkSize)))
Hannah Howard's avatar
Hannah Howard committed
734 735
	require.NoError(t, err, "unable to setup dag builder")

736
	nd, err := balanced.Layout(db)
Hannah Howard's avatar
Hannah Howard committed
737 738
	require.NoError(t, err, "unable to create unix fs node")

739
	err = bufferedDS.Commit()
Hannah Howard's avatar
Hannah Howard committed
740
	require.NoError(t, err, "unable to commit unix fs node")
741 742 743 744 745 746 747 748 749 750 751 752 753

	// save the original files bytes
	origBytes := buf.Bytes()

	// setup an IPLD loader/storer for blockstore 1
	loader1 := makeLoader(bs1)
	storer1 := makeStorer(bs1)

	// setup an IPLD loader/storer for blockstore 2
	loader2 := makeLoader(bs2)
	storer2 := makeStorer(bs2)

	td := newGsTestData(ctx, t)
754 755
	requestor := New(ctx, td.gsnet1, loader1, storer1)
	responder := New(ctx, td.gsnet2, loader2, storer2)
756
	extensionName := graphsync.ExtensionName("Free for all")
757
	responder.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
758 759 760 761 762 763
		hookActions.ValidateRequest()
		hookActions.SendExtensionData(graphsync.ExtensionData{
			Name: extensionName,
			Data: nil,
		})
	})
764

765 766 767 768
	// make a go-ipld-prime link for the root UnixFS node
	clink := cidlink.Link{Cid: nd.Cid()}

	// create a selector for the whole UnixFS dag
Hannah Howard's avatar
Hannah Howard committed
769
	ssb := builder.NewSelectorSpecBuilder(basicnode.Style.Any)
770 771 772 773 774 775 776 777 778 779 780 781

	allSelector := ssb.ExploreRecursive(ipldselector.RecursionLimitNone(),
		ssb.ExploreAll(ssb.ExploreRecursiveEdge())).Node()

	// execute the traversal
	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), clink, allSelector,
		graphsync.ExtensionData{
			Name: extensionName,
			Data: nil,
		})

	_ = testutil.CollectResponses(ctx, t, progressChan)
Hannah Howard's avatar
Hannah Howard committed
782
	testutil.VerifyEmptyErrors(ctx, t, errChan)
783 784 785 786 787 788

	// setup a DagService for the second block store
	dagService1 := merkledag.NewDAGService(blockservice.New(bs1, offline.Exchange(bs1)))

	// load the root of the UnixFS DAG from the new blockstore
	otherNode, err := dagService1.Get(ctx, nd.Cid())
Hannah Howard's avatar
Hannah Howard committed
789
	require.NoError(t, err, "should have been able to read received root node but didn't")
790 791 792

	// Setup a UnixFS file reader
	n, err := unixfile.NewUnixfsFile(ctx, dagService1, otherNode)
Hannah Howard's avatar
Hannah Howard committed
793
	require.NoError(t, err, "should have been able to setup UnixFS file but wasn't")
794 795

	fn, ok := n.(files.File)
Hannah Howard's avatar
Hannah Howard committed
796
	require.True(t, ok, "file should be a regular file, but wasn't")
797 798 799

	// Read the bytes for the UnixFS File
	finalBytes, err := ioutil.ReadAll(fn)
Hannah Howard's avatar
Hannah Howard committed
800
	require.NoError(t, err, "should have been able to read all of unix FS file but wasn't")
801 802

	// verify original bytes match final bytes!
Hannah Howard's avatar
Hannah Howard committed
803
	require.Equal(t, origBytes, finalBytes, "should have gotten same bytes written as read but didn't")
804 805
}

806 807 808 809 810 811 812 813 814 815 816 817 818 819 820
type gsTestData struct {
	mn                       mocknet.Mocknet
	ctx                      context.Context
	host1                    host.Host
	host2                    host.Host
	gsnet1                   gsnet.GraphSyncNetwork
	gsnet2                   gsnet.GraphSyncNetwork
	blockStore1, blockStore2 map[ipld.Link][]byte
	loader1, loader2         ipld.Loader
	storer1, storer2         ipld.Storer
	extensionData            []byte
	extensionName            graphsync.ExtensionName
	extension                graphsync.ExtensionData
	extensionResponseData    []byte
	extensionResponse        graphsync.ExtensionData
Hannah Howard's avatar
Hannah Howard committed
821 822
	extensionUpdateData      []byte
	extensionUpdate          graphsync.ExtensionData
823 824 825 826 827 828
}

func newGsTestData(ctx context.Context, t *testing.T) *gsTestData {
	td := &gsTestData{ctx: ctx}
	td.mn = mocknet.New(ctx)
	var err error
829
	// setup network
830
	td.host1, err = td.mn.GenPeer()
Hannah Howard's avatar
Hannah Howard committed
831
	require.NoError(t, err, "error generating host")
832
	td.host2, err = td.mn.GenPeer()
Hannah Howard's avatar
Hannah Howard committed
833
	require.NoError(t, err, "error generating host")
834
	err = td.mn.LinkAll()
Hannah Howard's avatar
Hannah Howard committed
835
	require.NoError(t, err, "error linking hosts")
836

837 838 839
	td.gsnet1 = gsnet.NewFromLibp2pHost(td.host1)
	td.gsnet2 = gsnet.NewFromLibp2pHost(td.host2)
	td.blockStore1 = make(map[ipld.Link][]byte)
840
	td.loader1, td.storer1 = testutil.NewTestStore(td.blockStore1)
841
	td.blockStore2 = make(map[ipld.Link][]byte)
842
	td.loader2, td.storer2 = testutil.NewTestStore(td.blockStore2)
843 844 845 846 847 848 849 850 851 852 853 854
	// setup extension handlers
	td.extensionData = testutil.RandomBytes(100)
	td.extensionName = graphsync.ExtensionName("AppleSauce/McGee")
	td.extension = graphsync.ExtensionData{
		Name: td.extensionName,
		Data: td.extensionData,
	}
	td.extensionResponseData = testutil.RandomBytes(100)
	td.extensionResponse = graphsync.ExtensionData{
		Name: td.extensionName,
		Data: td.extensionResponseData,
	}
Hannah Howard's avatar
Hannah Howard committed
855 856 857 858 859
	td.extensionUpdateData = testutil.RandomBytes(100)
	td.extensionUpdate = graphsync.ExtensionData{
		Name: td.extensionName,
		Data: td.extensionUpdateData,
	}
860

861 862
	return td
}
863

864 865
func (td *gsTestData) GraphSyncHost1(options ...Option) graphsync.GraphExchange {
	return New(td.ctx, td.gsnet1, td.loader1, td.storer1, options...)
866
}
867

868
func (td *gsTestData) GraphSyncHost2(options ...Option) graphsync.GraphExchange {
869

870
	return New(td.ctx, td.gsnet2, td.loader2, td.storer2, options...)
871
}
872

873 874 875 876
type receivedMessage struct {
	message gsmsg.GraphSyncMessage
	sender  peer.ID
}
877

878 879 880 881
// Receiver is an interface for receiving messages from the GraphSyncNetwork.
type receiver struct {
	messageReceived chan receivedMessage
}
882

883 884 885 886
func (r *receiver) ReceiveMessage(
	ctx context.Context,
	sender peer.ID,
	incoming gsmsg.GraphSyncMessage) {
887

888 889 890 891 892
	select {
	case <-ctx.Done():
	case r.messageReceived <- receivedMessage{incoming, sender}:
	}
}
893

894 895
func (r *receiver) ReceiveError(err error) {
}
896

897 898 899 900 901
func (r *receiver) Connected(p peer.ID) {
}

func (r *receiver) Disconnected(p peer.ID) {
}