fix(retrieval): resolve retrieval close event panic
Refactor ClientRetrieve command to remove the possibility of a send on close channel race condition
This commit is contained in:
parent
348dadf7ae
commit
10c1399474
@ -432,6 +432,45 @@ func (a *API) ClientRetrieveWithEvents(ctx context.Context, order api.RetrievalO
|
|||||||
return events, nil
|
return events, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type retrievalSubscribeEvent struct {
|
||||||
|
event rm.ClientEvent
|
||||||
|
state rm.ClientDealState
|
||||||
|
}
|
||||||
|
|
||||||
|
func readSubscribeEvents(ctx context.Context, subscribeEvents chan retrievalSubscribeEvent, events chan marketevents.RetrievalEvent) error {
|
||||||
|
for {
|
||||||
|
var subscribeEvent retrievalSubscribeEvent
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return xerrors.New("Retrieval Timed Out")
|
||||||
|
case subscribeEvent = <-subscribeEvents:
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return xerrors.New("Retrieval Timed Out")
|
||||||
|
case events <- marketevents.RetrievalEvent{
|
||||||
|
Event: subscribeEvent.event,
|
||||||
|
Status: subscribeEvent.state.Status,
|
||||||
|
BytesReceived: subscribeEvent.state.TotalReceived,
|
||||||
|
FundsSpent: subscribeEvent.state.FundsSpent,
|
||||||
|
}:
|
||||||
|
}
|
||||||
|
|
||||||
|
state := subscribeEvent.state
|
||||||
|
switch state.Status {
|
||||||
|
case rm.DealStatusCompleted:
|
||||||
|
return nil
|
||||||
|
case rm.DealStatusRejected:
|
||||||
|
return xerrors.Errorf("Retrieval Proposal Rejected: %s", state.Message)
|
||||||
|
case
|
||||||
|
rm.DealStatusDealNotFound,
|
||||||
|
rm.DealStatusErrored:
|
||||||
|
return xerrors.Errorf("Retrieval Error: %s", state.Message)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef, events chan marketevents.RetrievalEvent) {
|
func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref *api.FileRef, events chan marketevents.RetrievalEvent) {
|
||||||
defer close(events)
|
defer close(events)
|
||||||
|
|
||||||
@ -467,33 +506,13 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
|
|||||||
return err
|
return err
|
||||||
}*/
|
}*/
|
||||||
|
|
||||||
retrievalResult := make(chan error, 1)
|
var dealID retrievalmarket.DealID
|
||||||
|
subscribeEvents := make(chan retrievalSubscribeEvent, 1)
|
||||||
var dealId retrievalmarket.DealID
|
|
||||||
|
|
||||||
unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
|
unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
|
||||||
if state.PayloadCID.Equals(order.Root) && state.ID == dealId {
|
if state.PayloadCID.Equals(order.Root) && state.ID == dealID {
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
case subscribeEvents <- retrievalSubscribeEvent{event, state}:
|
||||||
case events <- marketevents.RetrievalEvent{
|
|
||||||
Event: event,
|
|
||||||
Status: state.Status,
|
|
||||||
BytesReceived: state.TotalReceived,
|
|
||||||
FundsSpent: state.FundsSpent,
|
|
||||||
}:
|
|
||||||
}
|
|
||||||
|
|
||||||
switch state.Status {
|
|
||||||
case rm.DealStatusCompleted:
|
|
||||||
retrievalResult <- nil
|
|
||||||
case rm.DealStatusRejected:
|
|
||||||
retrievalResult <- xerrors.Errorf("Retrieval Proposal Rejected: %s", state.Message)
|
|
||||||
case
|
|
||||||
rm.DealStatusDealNotFound,
|
|
||||||
rm.DealStatusErrored:
|
|
||||||
retrievalResult <- xerrors.Errorf("Retrieval Error: %s", state.Message)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@ -516,7 +535,7 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
|
|||||||
_ = a.RetrievalStoreMgr.ReleaseStore(store)
|
_ = a.RetrievalStoreMgr.ReleaseStore(store)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
dealId, err = a.Retrieval.Retrieve(
|
dealID, err = a.Retrieval.Retrieve(
|
||||||
ctx,
|
ctx,
|
||||||
order.Root,
|
order.Root,
|
||||||
params,
|
params,
|
||||||
@ -531,17 +550,12 @@ func (a *API) clientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
select {
|
err = readSubscribeEvents(ctx, subscribeEvents, events)
|
||||||
case <-ctx.Done():
|
|
||||||
unsubscribe()
|
unsubscribe()
|
||||||
finish(xerrors.New("Retrieval Timed Out"))
|
if err != nil {
|
||||||
|
finish(xerrors.Errorf("Retrieve: %w", err))
|
||||||
return
|
return
|
||||||
case err := <-retrievalResult:
|
|
||||||
unsubscribe()
|
|
||||||
if err != nil {
|
|
||||||
finish(xerrors.Errorf("Retrieve: %w", err))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// If ref is nil, it only fetches the data into the configured blockstore.
|
// If ref is nil, it only fetches the data into the configured blockstore.
|
||||||
|
Loading…
Reference in New Issue
Block a user