requestmanager.go 18.1 KB
Newer Older
1 2 3 4
package requestmanager

import (
	"context"
5
	"errors"
6
	"fmt"
Hannah Howard's avatar
Hannah Howard committed
7
	"sync/atomic"
8

Hannah Howard's avatar
Hannah Howard committed
9
	blocks "github.com/ipfs/go-block-format"
10
	"github.com/ipfs/go-cid"
Hannah Howard's avatar
Hannah Howard committed
11 12 13 14
	logging "github.com/ipfs/go-log"
	"github.com/ipld/go-ipld-prime"
	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
	"github.com/libp2p/go-libp2p-core/peer"
Hannah Howard's avatar
Hannah Howard committed
15

16
	"github.com/ipfs/go-graphsync"
Hannah Howard's avatar
Hannah Howard committed
17
	"github.com/ipfs/go-graphsync/cidset"
Hannah Howard's avatar
Hannah Howard committed
18
	"github.com/ipfs/go-graphsync/dedupkey"
19
	ipldutil "github.com/ipfs/go-graphsync/ipldutil"
20
	gsmsg "github.com/ipfs/go-graphsync/message"
21
	"github.com/ipfs/go-graphsync/metadata"
Hannah Howard's avatar
Hannah Howard committed
22 23
	"github.com/ipfs/go-graphsync/requestmanager/executor"
	"github.com/ipfs/go-graphsync/requestmanager/hooks"
24
	"github.com/ipfs/go-graphsync/requestmanager/types"
25 26
)

27
var log = logging.Logger("graphsync")
28 29

const (
30 31
	// defaultPriority is the default priority for requests sent by graphsync
	defaultPriority = graphsync.Priority(0)
32 33 34
)

type inProgressRequestStatus struct {
35 36 37 38 39 40 41 42
	ctx            context.Context
	cancelFn       func()
	p              peer.ID
	networkError   chan error
	resumeMessages chan []graphsync.ExtensionData
	pauseMessages  chan struct{}
	paused         bool
	lastResponse   atomic.Value
43 44 45 46
}

// PeerHandler is an interface that can send requests to peers
type PeerHandler interface {
47
	SendRequest(p peer.ID, graphSyncRequest gsmsg.GraphSyncRequest)
48 49
}

50 51 52
// AsyncLoader is an interface for loading links asynchronously, returning
// results as new responses are processed
type AsyncLoader interface {
53
	StartRequest(graphsync.RequestID, string) error
54
	ProcessResponse(responses map[graphsync.RequestID]metadata.Metadata,
55
		blks []blocks.Block)
56 57 58
	AsyncLoad(requestID graphsync.RequestID, link ipld.Link) <-chan types.AsyncLoadResult
	CompleteResponsesFor(requestID graphsync.RequestID)
	CleanupRequest(requestID graphsync.RequestID)
59 60
}

61 62 63 64 65 66 67 68
// RequestManager tracks outgoing requests and processes incoming reponses
// to them.
type RequestManager struct {
	ctx         context.Context
	cancel      func()
	messages    chan requestManagerMessage
	peerHandler PeerHandler
	rc          *responseCollector
69
	asyncLoader AsyncLoader
70
	// dont touch out side of run loop
71 72
	nextRequestID             graphsync.RequestID
	inProgressRequestStatuses map[graphsync.RequestID]*inProgressRequestStatus
Hannah Howard's avatar
Hannah Howard committed
73 74 75
	requestHooks              RequestHooks
	responseHooks             ResponseHooks
	blockHooks                BlockHooks
76 77 78 79 80 81
}

type requestManagerMessage interface {
	handle(rm *RequestManager)
}

Hannah Howard's avatar
Hannah Howard committed
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96
// RequestHooks run for new requests
type RequestHooks interface {
	ProcessRequestHooks(p peer.ID, request graphsync.RequestData) hooks.RequestResult
}

// ResponseHooks run for new responses
type ResponseHooks interface {
	ProcessResponseHooks(p peer.ID, response graphsync.ResponseData) hooks.UpdateResult
}

// BlockHooks run for each block loaded
type BlockHooks interface {
	ProcessBlockHooks(p peer.ID, response graphsync.ResponseData, block graphsync.BlockData) hooks.UpdateResult
}

97
// New generates a new request manager from a context, network, and selectorQuerier
Hannah Howard's avatar
Hannah Howard committed
98 99
func New(ctx context.Context,
	asyncLoader AsyncLoader,
Hannah Howard's avatar
Hannah Howard committed
100 101 102
	requestHooks RequestHooks,
	responseHooks ResponseHooks,
	blockHooks BlockHooks) *RequestManager {
103 104 105 106
	ctx, cancel := context.WithCancel(ctx)
	return &RequestManager{
		ctx:                       ctx,
		cancel:                    cancel,
107
		asyncLoader:               asyncLoader,
108 109
		rc:                        newResponseCollector(ctx),
		messages:                  make(chan requestManagerMessage, 16),
110
		inProgressRequestStatuses: make(map[graphsync.RequestID]*inProgressRequestStatus),
Hannah Howard's avatar
Hannah Howard committed
111 112
		requestHooks:              requestHooks,
		responseHooks:             responseHooks,
Hannah Howard's avatar
Hannah Howard committed
113
		blockHooks:                blockHooks,
114 115 116 117 118 119 120 121 122
	}
}

// SetDelegate specifies who will send messages out to the internet.
func (rm *RequestManager) SetDelegate(peerHandler PeerHandler) {
	rm.peerHandler = peerHandler
}

type inProgressRequest struct {
123 124
	requestID     graphsync.RequestID
	incoming      chan graphsync.ResponseProgress
125
	incomingError chan error
126 127 128 129
}

type newRequestMessage struct {
	p                     peer.ID
130
	root                  ipld.Link
131
	selector              ipld.Node
132
	extensions            []graphsync.ExtensionData
133 134 135 136
	inProgressRequestChan chan<- inProgressRequest
}

// SendRequest initiates a new GraphSync request to the given peer.
137 138
func (rm *RequestManager) SendRequest(ctx context.Context,
	p peer.ID,
139
	root ipld.Link,
140 141
	selector ipld.Node,
	extensions ...graphsync.ExtensionData) (<-chan graphsync.ResponseProgress, <-chan error) {
142
	if _, err := ipldutil.ParseSelector(selector); err != nil {
143
		return rm.singleErrorResponse(fmt.Errorf("Invalid Selector Spec"))
144 145 146 147 148
	}

	inProgressRequestChan := make(chan inProgressRequest)

	select {
149
	case rm.messages <- &newRequestMessage{p, root, selector, extensions, inProgressRequestChan}:
150
	case <-rm.ctx.Done():
151
		return rm.emptyResponse()
152
	case <-ctx.Done():
153
		return rm.emptyResponse()
154 155 156 157
	}
	var receivedInProgressRequest inProgressRequest
	select {
	case <-rm.ctx.Done():
158
		return rm.emptyResponse()
159 160 161
	case receivedInProgressRequest = <-inProgressRequestChan:
	}

162 163 164 165 166 167 168 169 170 171
	return rm.rc.collectResponses(ctx,
		receivedInProgressRequest.incoming,
		receivedInProgressRequest.incomingError,
		func() {
			rm.cancelRequest(receivedInProgressRequest.requestID,
				receivedInProgressRequest.incoming,
				receivedInProgressRequest.incomingError)
		})
}

172 173
func (rm *RequestManager) emptyResponse() (chan graphsync.ResponseProgress, chan error) {
	ch := make(chan graphsync.ResponseProgress)
174
	close(ch)
175
	errCh := make(chan error)
176 177 178 179
	close(errCh)
	return ch, errCh
}

180 181
func (rm *RequestManager) singleErrorResponse(err error) (chan graphsync.ResponseProgress, chan error) {
	ch := make(chan graphsync.ResponseProgress)
182
	close(ch)
183
	errCh := make(chan error, 1)
184
	errCh <- err
185 186
	close(errCh)
	return ch, errCh
187 188 189
}

type cancelRequestMessage struct {
190
	requestID graphsync.RequestID
191
	isPause   bool
192 193
}

194 195
func (rm *RequestManager) cancelRequest(requestID graphsync.RequestID,
	incomingResponses chan graphsync.ResponseProgress,
196
	incomingErrors chan error) {
197
	cancelMessageChannel := rm.messages
198
	for cancelMessageChannel != nil || incomingResponses != nil || incomingErrors != nil {
199
		select {
200
		case cancelMessageChannel <- &cancelRequestMessage{requestID, false}:
201 202 203 204 205
			cancelMessageChannel = nil
		// clear out any remaining responses, in case and "incoming reponse"
		// messages get processed before our cancel message
		case _, ok := <-incomingResponses:
			if !ok {
206 207 208 209 210
				incomingResponses = nil
			}
		case _, ok := <-incomingErrors:
			if !ok {
				incomingErrors = nil
211 212 213 214 215 216 217 218
			}
		case <-rm.ctx.Done():
			return
		}
	}
}

type processResponseMessage struct {
219
	p         peer.ID
220 221
	responses []gsmsg.GraphSyncResponse
	blks      []blocks.Block
222 223 224 225
}

// ProcessResponses ingests the given responses from the network and
// and updates the in progress requests based on those responses.
226
func (rm *RequestManager) ProcessResponses(p peer.ID, responses []gsmsg.GraphSyncResponse,
227
	blks []blocks.Block) {
228
	select {
229
	case rm.messages <- &processResponseMessage{p, responses, blks}:
230 231 232 233
	case <-rm.ctx.Done():
	}
}

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
type unpauseRequestMessage struct {
	id         graphsync.RequestID
	extensions []graphsync.ExtensionData
	response   chan error
}

// UnpauseRequest unpauses a request that was paused in a block hook based request ID
// Can also send extensions with unpause
func (rm *RequestManager) UnpauseRequest(requestID graphsync.RequestID, extensions ...graphsync.ExtensionData) error {
	response := make(chan error, 1)
	return rm.sendSyncMessage(&unpauseRequestMessage{requestID, extensions, response}, response)
}

type pauseRequestMessage struct {
	id       graphsync.RequestID
	response chan error
}

// PauseRequest pauses an in progress request (may take 1 or more blocks to process)
func (rm *RequestManager) PauseRequest(requestID graphsync.RequestID) error {
	response := make(chan error, 1)
	return rm.sendSyncMessage(&pauseRequestMessage{requestID, response}, response)
}

func (rm *RequestManager) sendSyncMessage(message requestManagerMessage, response chan error) error {
	select {
	case <-rm.ctx.Done():
		return errors.New("Context Cancelled")
	case rm.messages <- message:
	}
	select {
	case <-rm.ctx.Done():
		return errors.New("Context Cancelled")
	case err := <-response:
		return err
	}
}

272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298
// Startup starts processing for the WantManager.
func (rm *RequestManager) Startup() {
	go rm.run()
}

// Shutdown ends processing for the want manager.
func (rm *RequestManager) Shutdown() {
	rm.cancel()
}

func (rm *RequestManager) run() {
	// NOTE: Do not open any streams or connections from anywhere in this
	// event loop. Really, just don't do anything likely to block.
	defer rm.cleanupInProcessRequests()

	for {
		select {
		case message := <-rm.messages:
			message.handle(rm)
		case <-rm.ctx.Done():
			return
		}
	}
}

func (rm *RequestManager) cleanupInProcessRequests() {
	for _, requestStatus := range rm.inProgressRequestStatuses {
299
		requestStatus.cancelFn()
300 301 302
	}
}

303
type terminateRequestMessage struct {
304
	requestID graphsync.RequestID
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 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353
func (nrm *newRequestMessage) setupRequest(requestID graphsync.RequestID, rm *RequestManager) (chan graphsync.ResponseProgress, chan error) {
	request, hooksResult, err := rm.validateRequest(requestID, nrm.p, nrm.root, nrm.selector, nrm.extensions)
	if err != nil {
		return rm.singleErrorResponse(err)
	}
	doNotSendCidsData, has := request.Extension(graphsync.ExtensionDoNotSendCIDs)
	var doNotSendCids *cid.Set
	if has {
		doNotSendCids, err = cidset.DecodeCidSet(doNotSendCidsData)
		if err != nil {
			return rm.singleErrorResponse(err)
		}
	} else {
		doNotSendCids = cid.NewSet()
	}
	ctx, cancel := context.WithCancel(rm.ctx)
	p := nrm.p
	resumeMessages := make(chan []graphsync.ExtensionData, 1)
	pauseMessages := make(chan struct{}, 1)
	networkError := make(chan error, 1)
	requestStatus := &inProgressRequestStatus{
		ctx: ctx, cancelFn: cancel, p: p, resumeMessages: resumeMessages, pauseMessages: pauseMessages, networkError: networkError,
	}
	lastResponse := &requestStatus.lastResponse
	lastResponse.Store(gsmsg.NewResponse(request.ID(), graphsync.RequestAcknowledged))
	rm.inProgressRequestStatuses[request.ID()] = requestStatus
	incoming, incomingError := executor.ExecutionEnv{
		Ctx:              rm.ctx,
		SendRequest:      rm.peerHandler.SendRequest,
		TerminateRequest: rm.terminateRequest,
		RunBlockHooks:    rm.processBlockHooks,
		Loader:           rm.asyncLoader.AsyncLoad,
	}.Start(
		executor.RequestExecution{
			Ctx:              ctx,
			P:                p,
			Request:          request,
			NetworkError:     networkError,
			LastResponse:     lastResponse,
			DoNotSendCids:    doNotSendCids,
			NodeStyleChooser: hooksResult.CustomChooser,
			ResumeMessages:   resumeMessages,
			PauseMessages:    pauseMessages,
		})
	return incoming, incomingError
}

354
func (nrm *newRequestMessage) handle(rm *RequestManager) {
Hannah Howard's avatar
Hannah Howard committed
355 356
	var ipr inProgressRequest
	ipr.requestID = rm.nextRequestID
357
	rm.nextRequestID++
358
	ipr.incoming, ipr.incomingError = nrm.setupRequest(ipr.requestID, rm)
359 360

	select {
Hannah Howard's avatar
Hannah Howard committed
361
	case nrm.inProgressRequestChan <- ipr:
362 363 364 365
	case <-rm.ctx.Done():
	}
}

366 367 368 369 370
func (trm *terminateRequestMessage) handle(rm *RequestManager) {
	delete(rm.inProgressRequestStatuses, trm.requestID)
	rm.asyncLoader.CleanupRequest(trm.requestID)
}

371 372 373 374 375 376
func (crm *cancelRequestMessage) handle(rm *RequestManager) {
	inProgressRequestStatus, ok := rm.inProgressRequestStatuses[crm.requestID]
	if !ok {
		return
	}

377
	rm.peerHandler.SendRequest(inProgressRequestStatus.p, gsmsg.CancelRequest(crm.requestID))
378 379 380 381 382
	if crm.isPause {
		inProgressRequestStatus.paused = true
	} else {
		inProgressRequestStatus.cancelFn()
	}
383 384 385
}

func (prm *processResponseMessage) handle(rm *RequestManager) {
386 387
	filteredResponses := rm.processExtensions(prm.responses, prm.p)
	filteredResponses = rm.filterResponsesForPeer(filteredResponses, prm.p)
Hannah Howard's avatar
Hannah Howard committed
388
	rm.updateLastResponses(filteredResponses)
389
	responseMetadata := metadataForResponses(filteredResponses)
390 391 392 393 394 395 396 397 398 399
	rm.asyncLoader.ProcessResponse(responseMetadata, prm.blks)
	rm.processTerminations(filteredResponses)
}

func (rm *RequestManager) filterResponsesForPeer(responses []gsmsg.GraphSyncResponse, p peer.ID) []gsmsg.GraphSyncResponse {
	responsesForPeer := make([]gsmsg.GraphSyncResponse, 0, len(responses))
	for _, response := range responses {
		requestStatus, ok := rm.inProgressRequestStatuses[response.RequestID()]
		if !ok || requestStatus.p != p {
			continue
400
		}
401
		responsesForPeer = append(responsesForPeer, response)
402
	}
403 404
	return responsesForPeer
}
405

406 407 408 409 410 411 412 413 414 415 416
func (rm *RequestManager) processExtensions(responses []gsmsg.GraphSyncResponse, p peer.ID) []gsmsg.GraphSyncResponse {
	remainingResponses := make([]gsmsg.GraphSyncResponse, 0, len(responses))
	for _, response := range responses {
		success := rm.processExtensionsForResponse(p, response)
		if success {
			remainingResponses = append(remainingResponses, response)
		}
	}
	return remainingResponses
}

Hannah Howard's avatar
Hannah Howard committed
417 418 419 420 421 422
func (rm *RequestManager) updateLastResponses(responses []gsmsg.GraphSyncResponse) {
	for _, response := range responses {
		rm.inProgressRequestStatuses[response.RequestID()].lastResponse.Store(response)
	}
}

423
func (rm *RequestManager) processExtensionsForResponse(p peer.ID, response gsmsg.GraphSyncResponse) bool {
Hannah Howard's avatar
Hannah Howard committed
424 425 426 427 428 429
	result := rm.responseHooks.ProcessResponseHooks(p, response)
	if len(result.Extensions) > 0 {
		updateRequest := gsmsg.UpdateRequest(response.RequestID(), result.Extensions...)
		rm.peerHandler.SendRequest(p, updateRequest)
	}
	if result.Err != nil {
430 431 432 433
		requestStatus, ok := rm.inProgressRequestStatuses[response.RequestID()]
		if !ok {
			return false
		}
Hannah Howard's avatar
Hannah Howard committed
434 435 436 437
		responseError := rm.generateResponseErrorFromStatus(graphsync.RequestFailedUnknown)
		select {
		case requestStatus.networkError <- responseError:
		case <-requestStatus.ctx.Done():
438
		}
Hannah Howard's avatar
Hannah Howard committed
439
		rm.peerHandler.SendRequest(p, gsmsg.CancelRequest(response.RequestID()))
Hannah Howard's avatar
Hannah Howard committed
440 441
		requestStatus.cancelFn()
		return false
442 443 444 445
	}
	return true
}

446 447
func (rm *RequestManager) processTerminations(responses []gsmsg.GraphSyncResponse) {
	for _, response := range responses {
448
		if gsmsg.IsTerminalResponseCode(response.Status()) {
449 450 451 452 453 454
			if gsmsg.IsTerminalFailureCode(response.Status()) {
				requestStatus := rm.inProgressRequestStatuses[response.RequestID()]
				responseError := rm.generateResponseErrorFromStatus(response.Status())
				select {
				case requestStatus.networkError <- responseError:
				case <-requestStatus.ctx.Done():
455
				}
456
				requestStatus.cancelFn()
457
			}
458
			rm.asyncLoader.CompleteResponsesFor(response.RequestID())
459 460 461
		}
	}
}
462

463
func (rm *RequestManager) generateResponseErrorFromStatus(status graphsync.ResponseStatusCode) error {
464
	switch status {
465
	case graphsync.RequestFailedBusy:
466
		return graphsync.RequestFailedBusyErr{}
467
	case graphsync.RequestFailedContentNotFound:
468
		return graphsync.RequestFailedContentNotFoundErr{}
469
	case graphsync.RequestFailedLegal:
470
		return graphsync.RequestFailedLegalErr{}
471
	case graphsync.RequestFailedUnknown:
472 473 474
		return graphsync.RequestFailedUnknownErr{}
	case graphsync.RequestCancelled:
		return graphsync.RequestCancelledErr{}
475
	default:
476
		return fmt.Errorf("Unknown")
477 478
	}
}
479

Hannah Howard's avatar
Hannah Howard committed
480 481 482 483 484
func (rm *RequestManager) processBlockHooks(p peer.ID, response graphsync.ResponseData, block graphsync.BlockData) error {
	result := rm.blockHooks.ProcessBlockHooks(p, response, block)
	if len(result.Extensions) > 0 {
		updateRequest := gsmsg.UpdateRequest(response.RequestID(), result.Extensions...)
		rm.peerHandler.SendRequest(p, updateRequest)
485
	}
Hannah Howard's avatar
Hannah Howard committed
486
	if result.Err != nil {
487
		_, isPause := result.Err.(hooks.ErrPaused)
Hannah Howard's avatar
Hannah Howard committed
488 489
		select {
		case <-rm.ctx.Done():
490
		case rm.messages <- &cancelRequestMessage{response.RequestID(), isPause}:
Hannah Howard's avatar
Hannah Howard committed
491
		}
492
	}
Hannah Howard's avatar
Hannah Howard committed
493 494 495
	return result.Err
}

496 497 498 499 500 501 502
func (rm *RequestManager) terminateRequest(requestID graphsync.RequestID) {
	select {
	case <-rm.ctx.Done():
	case rm.messages <- &terminateRequestMessage{requestID}:
	}
}

503
func (rm *RequestManager) validateRequest(requestID graphsync.RequestID, p peer.ID, root ipld.Link, selectorSpec ipld.Node, extensions []graphsync.ExtensionData) (gsmsg.GraphSyncRequest, hooks.RequestResult, error) {
Hannah Howard's avatar
Hannah Howard committed
504 505
	_, err := ipldutil.EncodeNode(selectorSpec)
	if err != nil {
506
		return gsmsg.GraphSyncRequest{}, hooks.RequestResult{}, err
Hannah Howard's avatar
Hannah Howard committed
507
	}
508
	_, err = ipldutil.ParseSelector(selectorSpec)
Hannah Howard's avatar
Hannah Howard committed
509
	if err != nil {
510
		return gsmsg.GraphSyncRequest{}, hooks.RequestResult{}, err
Hannah Howard's avatar
Hannah Howard committed
511 512 513
	}
	asCidLink, ok := root.(cidlink.Link)
	if !ok {
514
		return gsmsg.GraphSyncRequest{}, hooks.RequestResult{}, fmt.Errorf("request failed: link has no cid")
Hannah Howard's avatar
Hannah Howard committed
515 516 517
	}
	request := gsmsg.NewRequest(requestID, asCidLink.Cid, selectorSpec, defaultPriority, extensions...)
	hooksResult := rm.requestHooks.ProcessRequestHooks(p, request)
Hannah Howard's avatar
Hannah Howard committed
518 519 520 521 522 523 524 525 526 527 528 529
	if hooksResult.PersistenceOption != "" {
		dedupData, err := dedupkey.EncodeDedupKey(hooksResult.PersistenceOption)
		if err != nil {
			return gsmsg.GraphSyncRequest{}, hooks.RequestResult{}, err
		}
		request = request.ReplaceExtensions([]graphsync.ExtensionData{
			{
				Name: graphsync.ExtensionDeDupByKey,
				Data: dedupData,
			},
		})
	}
Hannah Howard's avatar
Hannah Howard committed
530 531
	err = rm.asyncLoader.StartRequest(requestID, hooksResult.PersistenceOption)
	if err != nil {
532
		return gsmsg.GraphSyncRequest{}, hooks.RequestResult{}, err
Hannah Howard's avatar
Hannah Howard committed
533
	}
534
	return request, hooksResult, nil
Hannah Howard's avatar
Hannah Howard committed
535
}
536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 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

func (urm *unpauseRequestMessage) unpause(rm *RequestManager) error {
	inProgressRequestStatus, ok := rm.inProgressRequestStatuses[urm.id]
	if !ok {
		return errors.New("request not found")
	}
	if !inProgressRequestStatus.paused {
		return errors.New("request is not paused")
	}
	inProgressRequestStatus.paused = false
	select {
	case <-inProgressRequestStatus.pauseMessages:
		rm.peerHandler.SendRequest(inProgressRequestStatus.p, gsmsg.UpdateRequest(urm.id, urm.extensions...))
		return nil
	case <-rm.ctx.Done():
		return errors.New("context cancelled")
	case inProgressRequestStatus.resumeMessages <- urm.extensions:
		return nil
	}
}
func (urm *unpauseRequestMessage) handle(rm *RequestManager) {
	err := urm.unpause(rm)
	select {
	case <-rm.ctx.Done():
	case urm.response <- err:
	}
}
func (prm *pauseRequestMessage) pause(rm *RequestManager) error {
	inProgressRequestStatus, ok := rm.inProgressRequestStatuses[prm.id]
	if !ok {
		return errors.New("request not found")
	}
	if inProgressRequestStatus.paused {
		return errors.New("request is already paused")
	}
	inProgressRequestStatus.paused = true
	select {
	case <-rm.ctx.Done():
		return errors.New("context cancelled")
	case inProgressRequestStatus.pauseMessages <- struct{}{}:
		return nil
	}
}
func (prm *pauseRequestMessage) handle(rm *RequestManager) {
	err := prm.pause(rm)
	select {
	case <-rm.ctx.Done():
	case prm.response <- err:
	}
}