Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Don't run hooks on blocks we didn't have #331

Merged
merged 2 commits into from
Jan 13, 2022
Merged

Don't run hooks on blocks we didn't have #331

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
21cd7bd
fix(executor): don't run hooks on blocks we didn't have
hannahhoward Jan 13, 2022
5727192
fix(executor): fix test
hannahhoward Jan 13, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion requestmanager/executor/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,7 @@ func (e *Executor) traverse(rt RequestTask) error {
if err != nil {
return err
}

}
}

Expand Down Expand Up @@ -201,7 +202,10 @@ func (e *Executor) advanceTraversal(rt RequestTask, result types.AsyncLoadResult
}

func (e *Executor) processResult(rt RequestTask, link ipld.Link, result types.AsyncLoadResult) error {
err := e.onNewBlock(rt, &blockData{link, result.Local, uint64(len(result.Data)), int64(rt.Traverser.NBlocksTraversed())})
var err error
if result.Err == nil {
err = e.onNewBlock(rt, &blockData{link, result.Local, uint64(len(result.Data)), int64(rt.Traverser.NBlocksTraversed())})
}
select {
case <-rt.PauseMessages:
if err == nil {
Expand Down
67 changes: 40 additions & 27 deletions requestmanager/executor/executor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,8 @@ import (
"github.com/ipfs/go-graphsync/testutil"
)

type configureLoaderFn func(p peer.ID, requestID graphsync.RequestID, tbc *testutil.TestBlockChain, fal *testloader.FakeAsyncLoader, startStop [2]int)

func TestRequestExecutionBlockChain(t *testing.T) {
testCases := map[string]struct {
configureLoader configureLoaderFn
configureRequestExecution func(p peer.ID, requestID graphsync.RequestID, tbc *testutil.TestBlockChain, ree *requestExecutionEnv)
verifyResults func(t *testing.T, tbc *testutil.TestBlockChain, ree *requestExecutionEnv, responses []graphsync.ResponseProgress, receivedErrors []error)
}{
Expand All @@ -44,6 +41,26 @@ func TestRequestExecutionBlockChain(t *testing.T) {
require.NoError(t, ree.terminalError)
},
},
"missing block case": {
configureRequestExecution: func(p peer.ID, requestID graphsync.RequestID, tbc *testutil.TestBlockChain, ree *requestExecutionEnv) {
ree.customRemoteBehavior = func() {
// pretend the remote sent five blocks before encountering a missing block
ree.fal.SuccessResponseOn(p, requestID, tbc.Blocks(0, 5))
missingCid := cidlink.Link{Cid: tbc.Blocks(5, 6)[0].Cid()}
ree.fal.ResponseOn(p, requestID, missingCid, types.AsyncLoadResult{Err: graphsync.RemoteMissingBlockErr{Link: missingCid}})
}
},
verifyResults: func(t *testing.T, tbc *testutil.TestBlockChain, ree *requestExecutionEnv, responses []graphsync.ResponseProgress, receivedErrors []error) {
tbc.VerifyResponseRangeSync(responses, 0, 5)
require.Len(t, receivedErrors, 1)
require.Equal(t, receivedErrors[0], graphsync.RemoteMissingBlockErr{Link: cidlink.Link{Cid: tbc.Blocks(5, 6)[0].Cid()}})
require.Equal(t, []requestSent{{ree.p, ree.request}}, ree.requestsSent)
// we should only call block hooks for blocks we actually received
require.Len(t, ree.blookHooksCalled, 5)
require.NoError(t, ree.terminalError)
},
},

"error at block hook": {
configureRequestExecution: func(p peer.ID, requestID graphsync.RequestID, tbc *testutil.TestBlockChain, ree *requestExecutionEnv) {
ree.blockHookResults[blockHookKey{p, requestID, tbc.LinkTipIndex(5)}] = hooks.UpdateResult{Err: errors.New("something went wrong")}
Expand Down Expand Up @@ -181,12 +198,6 @@ func TestRequestExecutionBlockChain(t *testing.T) {
fal := testloader.NewFakeAsyncLoader()
requestID := graphsync.RequestID(rand.Int31())
p := testutil.GeneratePeers(1)[0]
configureLoader := data.configureLoader
if configureLoader == nil {
configureLoader = func(p peer.ID, requestID graphsync.RequestID, tbc *testutil.TestBlockChain, fal *testloader.FakeAsyncLoader, startStop [2]int) {
fal.SuccessResponseOn(p, requestID, tbc.Blocks(startStop[0], startStop[1]))
}
}
requestCtx, requestCancel := context.WithCancel(ctx)
defer requestCancel()
var responsesReceived []graphsync.ResponseProgress
Expand All @@ -199,7 +210,6 @@ func TestRequestExecutionBlockChain(t *testing.T) {
request: gsmsg.NewRequest(requestID, tbc.TipLink.(cidlink.Link).Cid, tbc.Selector(), graphsync.Priority(rand.Int31())),
fal: fal,
tbc: tbc,
configureLoader: configureLoader,
initialRequest: true,
inProgressErr: make(chan error, 1),
traverser: ipldutil.TraversalBuilder{
Expand All @@ -219,7 +229,7 @@ func TestRequestExecutionBlockChain(t *testing.T) {
if data.configureRequestExecution != nil {
data.configureRequestExecution(p, requestID, tbc, ree)
}
ree.configureLoader(p, requestID, tbc, fal, [2]int{0, ree.loadLocallyUntil})
ree.fal.SuccessResponseOn(p, requestID, tbc.Blocks(0, ree.loadLocallyUntil))
var errorsReceived []error
errCollectionErr := make(chan error, 1)
go func() {
Expand Down Expand Up @@ -262,27 +272,26 @@ type pauseKey struct {

type requestExecutionEnv struct {
// params
ctx context.Context
request gsmsg.GraphSyncRequest
p peer.ID
blockHookResults map[blockHookKey]hooks.UpdateResult
doNotSendCids *cid.Set
pauseMessages chan struct{}
externalPause pauseKey
loadLocallyUntil int
traverser ipldutil.Traverser
inProgressErr chan error
initialRequest bool

ctx context.Context
request gsmsg.GraphSyncRequest
p peer.ID
blockHookResults map[blockHookKey]hooks.UpdateResult
doNotSendCids *cid.Set
pauseMessages chan struct{}
externalPause pauseKey
loadLocallyUntil int
traverser ipldutil.Traverser
inProgressErr chan error
initialRequest bool
customRemoteBehavior func()
// results
requestsSent []requestSent
blookHooksCalled []blockHookKey
terminalError error

// deps
configureLoader configureLoaderFn
tbc *testutil.TestBlockChain
fal *testloader.FakeAsyncLoader
tbc *testutil.TestBlockChain
fal *testloader.FakeAsyncLoader
}

func (ree *requestExecutionEnv) ReleaseRequestTask(_ peer.ID, _ *peertask.Task, err error) {
Expand Down Expand Up @@ -317,7 +326,11 @@ func (ree *requestExecutionEnv) GetRequestTask(_ peer.ID, _ *peertask.Task, requ
func (ree *requestExecutionEnv) SendRequest(p peer.ID, request gsmsg.GraphSyncRequest) {
ree.requestsSent = append(ree.requestsSent, requestSent{p, request})
if !request.IsCancel() && !request.IsUpdate() {
ree.configureLoader(ree.p, ree.request.ID(), ree.tbc, ree.fal, [2]int{ree.loadLocallyUntil, len(ree.tbc.AllBlocks())})
if ree.customRemoteBehavior == nil {
ree.fal.SuccessResponseOn(p, request.ID(), ree.tbc.Blocks(ree.loadLocallyUntil, len(ree.tbc.AllBlocks())))
} else {
ree.customRemoteBehavior()
}
}
}

Expand Down