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.

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

feat(routing/http/server): add routing timeout #720

Merged
merged 3 commits into from
Nov 19, 2024
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
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
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ The following emojis are used to highlight certain changes:
### Added

- `routing/http/server`: added Prometheus instrumentation to http delegated routing endpoints.
- `routing/http/server`: added a routing timeout with default value of 30 seconds to delegated routing server to avoid indefinite hanging on content/peer routing requests.
lidel marked this conversation as resolved.
Show resolved Hide resolved

### Changed

Expand Down
30 changes: 26 additions & 4 deletions routing/http/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@

DefaultRecordsLimit = 20
DefaultStreamingRecordsLimit = 0
DefaultRoutingTimeout = 30 * time.Second
)

var logger = logging.Logger("routing/http/server")
Expand Down Expand Up @@ -132,11 +133,18 @@
}
}

func WithRoutingTimeout(timeout time.Duration) Option {
return func(s *server) {
s.routingTimeout = timeout
}

Check warning on line 139 in routing/http/server/server.go

View check run for this annotation

Codecov / codecov/patch

routing/http/server/server.go#L136-L139

Added lines #L136 - L139 were not covered by tests
}

func Handler(svc ContentRouter, opts ...Option) http.Handler {
server := &server{
svc: svc,
recordsLimit: DefaultRecordsLimit,
streamingRecordsLimit: DefaultStreamingRecordsLimit,
routingTimeout: DefaultRoutingTimeout,
}

for _, opt := range opts {
Expand Down Expand Up @@ -174,6 +182,7 @@
recordsLimit int
streamingRecordsLimit int
promRegistry prometheus.Registerer
routingTimeout time.Duration
}

func (s *server) detectResponseType(r *http.Request) (string, error) {
Expand Down Expand Up @@ -246,7 +255,10 @@
recordsLimit = s.recordsLimit
}

provIter, err := s.svc.FindProviders(httpReq.Context(), cid, recordsLimit)
ctx, cancel := context.WithTimeout(httpReq.Context(), s.routingTimeout)
defer cancel()

provIter, err := s.svc.FindProviders(ctx, cid, recordsLimit)
if err != nil {
if errors.Is(err, routing.ErrNotFound) {
// handlerFunc takes care of setting the 404 and necessary headers
Expand Down Expand Up @@ -335,7 +347,11 @@
recordsLimit = s.recordsLimit
}

provIter, err := s.svc.FindPeers(r.Context(), pid, recordsLimit)
// Add timeout to the routing operation
ctx, cancel := context.WithTimeout(r.Context(), s.routingTimeout)
defer cancel()

provIter, err := s.svc.FindPeers(ctx, pid, recordsLimit)
if err != nil {
if errors.Is(err, routing.ErrNotFound) {
// handlerFunc takes care of setting the 404 and necessary headers
Expand Down Expand Up @@ -466,7 +482,10 @@
return
}

record, err := s.svc.GetIPNS(r.Context(), name)
ctx, cancel := context.WithTimeout(r.Context(), s.routingTimeout)
defer cancel()

record, err := s.svc.GetIPNS(ctx, name)
if err != nil {
if errors.Is(err, routing.ErrNotFound) {
writeErr(w, "GetIPNS", http.StatusNotFound, fmt.Errorf("delegate error: %w", err))
Expand Down Expand Up @@ -550,7 +569,10 @@
return
}

err = s.svc.PutIPNS(r.Context(), name, record)
ctx, cancel := context.WithTimeout(r.Context(), s.routingTimeout)
defer cancel()

err = s.svc.PutIPNS(ctx, name, record)
if err != nil {
writeErr(w, "PutIPNS", http.StatusInternalServerError, fmt.Errorf("delegate error: %w", err))
return
Expand Down
Loading