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

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

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

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

21
	blocks "github.com/ipfs/go-block-format"
22
	"github.com/ipfs/go-blockservice"
23
	"github.com/ipfs/go-cid"
24 25 26 27 28 29 30 31 32 33 34 35
	"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"

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

38
	"github.com/ipfs/go-graphsync/cidset"
39
	"github.com/ipfs/go-graphsync/ipldutil"
40 41 42 43
	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"
Edgar Lee's avatar
Edgar Lee committed
44
	ipldselector "github.com/ipld/go-ipld-prime/traversal/selector"
45
	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
46
	"github.com/libp2p/go-libp2p-core/host"
47
	"github.com/libp2p/go-libp2p-core/peer"
48 49 50 51 52 53 54 55
	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()
56
	td := newGsTestData(ctx, t)
57
	r := &receiver{
58
		messageReceived: make(chan receivedMessage),
59
	}
60 61
	td.gsnet2.SetDelegate(r)
	graphSync := td.GraphSyncHost1()
62

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

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

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

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

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

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

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

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

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

116
	requestID := graphsync.RequestID(rand.Int31())
117 118

	message := gsmsg.New()
119
	message.AddRequest(gsmsg.NewRequest(requestID, blockChain.TipLink.(cidlink.Link).Cid, blockChain.Selector(), graphsync.Priority(math.MaxInt32), td.extension))
120
	// send request across network
121
	err := td.gsnet1.SendMessage(ctx, td.host2.ID(), message)
Hannah Howard's avatar
Hannah Howard committed
122
	require.NoError(t, err)
123 124 125
	// read the values sent back to requestor
	var received gsmsg.GraphSyncMessage
	var receivedBlocks []blocks.Block
126
	var receivedExtensions [][]byte
127
	for {
Hannah Howard's avatar
Hannah Howard committed
128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144
		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
145 146 147
		}
	}

Hannah Howard's avatar
Hannah Howard committed
148 149 150 151
	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")
152
}
153

154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174
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)
}

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

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

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

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

	var receivedResponseData []byte
	var receivedRequestData []byte

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

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

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

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

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

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

236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284
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)
}

285 286 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 336
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")

}

Hannah Howard's avatar
Hannah Howard committed
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
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
380 381 382 383 384 385 386 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 433 434 435 436 437 438 439 440
		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
441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456
		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")
}

457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495
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
496
			hookActions.UseLinkTargetNodeStyleChooser(blockChain.Chooser)
497 498 499 500 501 502
			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
503
			hookActions.UseLinkTargetNodeStyleChooser(blockChain.Chooser)
504 505 506 507 508 509
			hookActions.UsePersistenceOption("chainstore")
		}
	})

	progressChan, errChan := requestor.Request(ctx, td.host2.ID(), blockChain.TipLink, blockChain.Selector())
	testutil.VerifyEmptyResponse(ctx, t, progressChan)
510
	testutil.VerifyHasErrors(ctx, t, errChan)
511 512 513 514 515 516 517 518 519

	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")
}

520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535
// 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()
536 537 538 539 540
	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()
541

542 543
	// setup receiving peer to just record message coming in
	blockChainLength := 40
544
	blockChain := testutil.SetupBlockChain(ctx, t, td.loader1, td.storer2, 200000, blockChainLength)
545 546 547 548

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

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

Hannah Howard's avatar
Hannah Howard committed
551 552
	blockChain.VerifyWholeChain(ctx, progressChan)
	testutil.VerifyEmptyErrors(ctx, t, errChan)
553 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 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616
// 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
617
	require.NoError(t, err, "unable to create path for fixture file")
618 619

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

622 623 624 625 626 627 628 629 630 631 632 633 634 635 636
	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
637 638
	require.NoError(t, err, "unable to setup dag builder")

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

642
	err = bufferedDS.Commit()
Hannah Howard's avatar
Hannah Howard committed
643
	require.NoError(t, err, "unable to commit unix fs node")
644 645 646 647 648 649 650 651 652 653 654 655 656

	// 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)
657 658
	requestor := New(ctx, td.gsnet1, loader1, storer1)
	responder := New(ctx, td.gsnet2, loader2, storer2)
659
	extensionName := graphsync.ExtensionName("Free for all")
660
	responder.RegisterIncomingRequestHook(func(p peer.ID, requestData graphsync.RequestData, hookActions graphsync.IncomingRequestHookActions) {
661 662 663 664 665 666
		hookActions.ValidateRequest()
		hookActions.SendExtensionData(graphsync.ExtensionData{
			Name: extensionName,
			Data: nil,
		})
	})
667

668 669 670 671
	// 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
672
	ssb := builder.NewSelectorSpecBuilder(basicnode.Style.Any)
673 674 675 676 677 678 679 680 681 682 683 684

	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
685
	testutil.VerifyEmptyErrors(ctx, t, errChan)
686 687 688 689 690 691

	// 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
692
	require.NoError(t, err, "should have been able to read received root node but didn't")
693 694 695

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

	fn, ok := n.(files.File)
Hannah Howard's avatar
Hannah Howard committed
699
	require.True(t, ok, "file should be a regular file, but wasn't")
700 701 702

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

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

709 710 711 712 713 714 715 716 717 718 719 720 721 722 723
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
724 725
	extensionUpdateData      []byte
	extensionUpdate          graphsync.ExtensionData
726 727 728 729 730 731
}

func newGsTestData(ctx context.Context, t *testing.T) *gsTestData {
	td := &gsTestData{ctx: ctx}
	td.mn = mocknet.New(ctx)
	var err error
732
	// setup network
733
	td.host1, err = td.mn.GenPeer()
Hannah Howard's avatar
Hannah Howard committed
734
	require.NoError(t, err, "error generating host")
735
	td.host2, err = td.mn.GenPeer()
Hannah Howard's avatar
Hannah Howard committed
736
	require.NoError(t, err, "error generating host")
737
	err = td.mn.LinkAll()
Hannah Howard's avatar
Hannah Howard committed
738
	require.NoError(t, err, "error linking hosts")
739

740 741 742
	td.gsnet1 = gsnet.NewFromLibp2pHost(td.host1)
	td.gsnet2 = gsnet.NewFromLibp2pHost(td.host2)
	td.blockStore1 = make(map[ipld.Link][]byte)
743
	td.loader1, td.storer1 = testutil.NewTestStore(td.blockStore1)
744
	td.blockStore2 = make(map[ipld.Link][]byte)
745
	td.loader2, td.storer2 = testutil.NewTestStore(td.blockStore2)
746 747 748 749 750 751 752 753 754 755 756 757
	// 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
758 759 760 761 762
	td.extensionUpdateData = testutil.RandomBytes(100)
	td.extensionUpdate = graphsync.ExtensionData{
		Name: td.extensionName,
		Data: td.extensionUpdateData,
	}
763

764 765
	return td
}
766

767 768
func (td *gsTestData) GraphSyncHost1(options ...Option) graphsync.GraphExchange {
	return New(td.ctx, td.gsnet1, td.loader1, td.storer1, options...)
769
}
770

771
func (td *gsTestData) GraphSyncHost2(options ...Option) graphsync.GraphExchange {
772

773
	return New(td.ctx, td.gsnet2, td.loader2, td.storer2, options...)
774
}
775

776 777 778 779
type receivedMessage struct {
	message gsmsg.GraphSyncMessage
	sender  peer.ID
}
780

781 782 783 784
// Receiver is an interface for receiving messages from the GraphSyncNetwork.
type receiver struct {
	messageReceived chan receivedMessage
}
785

786 787 788 789
func (r *receiver) ReceiveMessage(
	ctx context.Context,
	sender peer.ID,
	incoming gsmsg.GraphSyncMessage) {
790

791 792 793 794 795
	select {
	case <-ctx.Done():
	case r.messageReceived <- receivedMessage{incoming, sender}:
	}
}
796

797 798
func (r *receiver) ReceiveError(err error) {
}
799

800 801 802 803 804
func (r *receiver) Connected(p peer.ID) {
}

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