Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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
9 changes: 9 additions & 0 deletions agent/internal/callable/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"mime"
"net/http"
Expand Down Expand Up @@ -114,6 +115,8 @@ func Via(r *http.Request) string {
return via
}

var errCallerGone = &Error{Code: "cancelled", Message: "CANCELLED"}

// callInfo is what the request log records about a call.
type callInfo struct {
provider string
Expand All @@ -123,6 +126,12 @@ type callInfo struct {
// serve writes the response and returns the error it sent, if any.
func (s *Server) serve(w http.ResponseWriter, r *http.Request, requireAuth bool, handler Handler) (call callInfo, err error) {
fail := func(e error) (callInfo, error) {
// A disconnecting caller cancels every downstream call, and handlers may
// wrap the resulting failure as a provider outage. The request context is
// the ground truth: if the caller left, report the call as cancelled.
if errors.Is(r.Context().Err(), context.Canceled) {
e = fmt.Errorf("%w: %w", errCallerGone, e)
}
writeError(w, e)
return call, e
}
Expand Down
21 changes: 21 additions & 0 deletions agent/internal/callable/server_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -225,3 +225,24 @@ func TestContextErrorsUseCanonicalStatus(t *testing.T) {
})
}
}

// Handlers may wrap a caller's disconnect as an explicit provider error; the
// request context, not the returned error, decides whether the caller left.
func TestCallerDisconnectOverridesExplicitError(t *testing.T) {
core, logs := observer.New(zap.InfoLevel)
defer zap.ReplaceGlobals(zap.New(observability.Wrap(core)))()
ctx, disconnect := context.WithCancel(context.Background())
server := NewServer(nil)
server.Register("getEmotes", false, func(ctx context.Context, _ json.RawMessage) (any, error) {
disconnect()
<-ctx.Done()
return nil, fmt.Errorf("twitch access token: %w: %w", &Error{Code: "unavailable", Message: "retry"}, status.Error(codes.Canceled, "context canceled"))
})
request := httptest.NewRequestWithContext(ctx, "POST", "/getEmotes", strings.NewReader(`{"data":{}}`))
request.Header.Set("Content-Type", "application/json")
server.ServeHTTP(httptest.NewRecorder(), request)
entries := logs.FilterMessage("http request").All()
if len(entries) != 1 || entries[0].Level != zap.InfoLevel || entries[0].ContextMap()["status"] != int64(499) || entries[0].ContextMap()["@type"] != nil {
t.Fatalf("caller disconnect reported as agent failure: %v", entries)
}
}
37 changes: 37 additions & 0 deletions agent/internal/catalog/error_context_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"errors"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
Expand All @@ -13,6 +14,8 @@ import (
"github.com/muxable/rtchat/agent/internal/clients"
"go.uber.org/zap"
"go.uber.org/zap/zaptest/observer"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)

func TestEmoteProviderFailureKeepsLogContextAndGenericResponse(t *testing.T) {
Expand Down Expand Up @@ -121,3 +124,37 @@ func TestEmotesRecoverFromTransientProviderError(t *testing.T) {
}
})
}

// A caller that disconnects cancels the token read; Firestore reports that as a
// gRPC Canceled status, which must not surface as a provider outage alert.
type disconnectTokens struct {
Tokens
disconnect context.CancelFunc
}

func (f disconnectTokens) AccessToken(ctx context.Context, _, _ string) (string, error) {
f.disconnect()
<-ctx.Done()
return "", status.Error(codes.Canceled, "context canceled")
}

func TestClientDisconnectIsNotProviderOutage(t *testing.T) {
s, _, _ := fixture()
api := callable.NewServer(func(context.Context, string) (string, error) { return "uid", nil })
ctx, disconnect := context.WithCancel(context.Background())
s.Tokens = disconnectTokens{Tokens: s.Tokens, disconnect: disconnect}
s.Register(api)
core, logs := observer.New(zap.InfoLevel)
defer zap.ReplaceGlobals(zap.New(core))()
r := httptest.NewRequestWithContext(ctx, "POST", "/getEmotes", strings.NewReader(`{"data":{"provider":"twitch","channelId":"123"}}`))
r.Header.Set("Content-Type", "application/json")
r.Header.Set("Authorization", "Bearer firebase-id-token")
api.ServeHTTP(httptest.NewRecorder(), r)
entries := logs.FilterMessage("http request").All()
if len(entries) != 1 {
t.Fatalf("got %d request logs", len(entries))
}
if fields := entries[0].ContextMap(); entries[0].Level != zap.InfoLevel || fields["status"] != int64(499) || fields["code"] != "cancelled" {
t.Fatalf("client disconnect reported as agent failure: level=%v %v", entries[0].Level, fields)
}
}
Loading