Repository navigation
Add channel health monitoring and timeout recovery controls #55
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
Merged
Changes from all commits
Commits
Show all changes
12 commits
Select commit
Hold shift + click to select a range
6060f99
update
CSCITech 7cb6a7a
fix: close channel health review findings
CSCITech f0b8fd4
fix: harden channel health recovery and routing
CSCITech d822eb6
fix: close channel health follow-up review findings
CSCITech 029c5cd
fix: close channel health review follow-ups
CSCITech df192ff
fix: exclude channel tests from health penalties
CSCITech 0bfd0b5
fix: close final channel health review findings
CSCITech a688f16
fix: harden channel health runtime projections
CSCITech 501739b
fix: close channel health review findings
CSCITech 4c03404
fix: close final channel health review findings
CSCITech 6b3e6e6
fix: harden channel timeout retries and caches
CSCITech 4ae54f6
fix: batch stale channel health cleanup
CSCITech File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,155 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "strconv" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/go-redis/redis/v8" | ||
| ) | ||
|
|
||
| const ( | ||
| channelHealthRuntimeKeyPrefix = "maxapi:channel-timeout:v1:observation:" | ||
| channelHealthRuntimeTTL = 24 * time.Hour | ||
| channelHealthObservationTimeout = 100 * time.Millisecond | ||
| channelHealthObservationQueueSize = 256 | ||
| ) | ||
|
|
||
| var channelHealthObservationQueue = make(chan ChannelHealthObservation, channelHealthObservationQueueSize) | ||
|
|
||
| var channelHealthObservationWorkerOnce sync.Once | ||
|
|
||
| func startChannelHealthObservationWorker() { | ||
| channelHealthObservationWorkerOnce.Do(func() { | ||
| go func() { | ||
| for observation := range channelHealthObservationQueue { | ||
| // The worker is deliberately best-effort. A slow or unavailable | ||
| // Redis instance must not remain on a relay response path. | ||
| _ = RecordChannelHealthObservation(observation) | ||
| } | ||
| }() | ||
| }) | ||
| } | ||
|
|
||
| // EnqueueChannelHealthObservation schedules a best-effort diagnostic write. | ||
| // It never waits for Redis and drops the observation when the bounded queue is | ||
| // full, preserving the upstream response path as fail-open. | ||
| func EnqueueChannelHealthObservation(observation ChannelHealthObservation) { | ||
| if !observation.PolicyEnabled || observation.ChannelID <= 0 || observation.Event == "" || !RedisEnabled || RDB == nil { | ||
| return | ||
| } | ||
| startChannelHealthObservationWorker() | ||
| select { | ||
| case channelHealthObservationQueue <- observation: | ||
| default: | ||
| // A full diagnostics queue is equivalent to an unavailable optional | ||
| // projection. The request path remains unaffected. | ||
| } | ||
| } | ||
|
|
||
| // ChannelHealthObservation is the minimal, non-sensitive runtime fact emitted | ||
| // by a selected upstream channel attempt. It deliberately carries no request | ||
| // or response body and no provider credential. | ||
| type ChannelHealthObservation struct { | ||
| ChannelID int | ||
| Event string | ||
| PolicyEnabled bool | ||
| ObservedAt time.Time | ||
| AttemptLatencyMS int64 | ||
| Stream bool | ||
| RetryIndex int | ||
| } | ||
|
|
||
| // ChannelHealthRuntimeTransition is a small process-local bridge from the | ||
| // relay/runtime package to Smart Operations. The runtime state itself remains | ||
| // in Redis; this hook only lets the existing notification projector react | ||
| // immediately without introducing a package cycle. | ||
| type ChannelHealthRuntimeTransition struct { | ||
| ChannelID int | ||
| Transition string | ||
| Penalty int64 | ||
| PenaltyUntil time.Time | ||
| RuntimeDisabled bool | ||
| DisabledUntil time.Time | ||
| ObservedAt time.Time | ||
| } | ||
|
|
||
| var channelHealthRuntimeTransitionObserver struct { | ||
| sync.RWMutex | ||
| fn func(ChannelHealthRuntimeTransition) | ||
| } | ||
|
|
||
| func SetChannelHealthRuntimeTransitionObserver(observer func(ChannelHealthRuntimeTransition)) { | ||
| channelHealthRuntimeTransitionObserver.Lock() | ||
| channelHealthRuntimeTransitionObserver.fn = observer | ||
| channelHealthRuntimeTransitionObserver.Unlock() | ||
| } | ||
|
|
||
| func PublishChannelHealthRuntimeTransition(transition ChannelHealthRuntimeTransition) { | ||
| channelHealthRuntimeTransitionObserver.RLock() | ||
| observer := channelHealthRuntimeTransitionObserver.fn | ||
| channelHealthRuntimeTransitionObserver.RUnlock() | ||
| if observer != nil { | ||
| observer(transition) | ||
| } | ||
| } | ||
|
|
||
| func isChannelHealthTimeoutEvent(event string) bool { | ||
| switch event { | ||
| case "transport_timeout", "stream_idle_timeout", "request_timeout", | ||
| "streaming_first_result_timeout", "non_streaming_response_timeout": | ||
| return true | ||
| default: | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| func channelHealthRuntimeKey(channelID int) string { | ||
| return fmt.Sprintf("%s%d", channelHealthRuntimeKeyPrefix, channelID) | ||
| } | ||
|
|
||
| // RecordChannelHealthObservation writes a bounded runtime projection. Redis | ||
| // failure is returned to the caller, which must treat it as diagnostic-only | ||
| // and never let it change relay behavior. | ||
| func RecordChannelHealthObservation(observation ChannelHealthObservation) error { | ||
| if observation.ChannelID <= 0 || observation.Event == "" { | ||
| return errors.New("channel health observation requires channel id and event") | ||
| } | ||
| if !RedisEnabled || RDB == nil { | ||
| return nil | ||
| } | ||
| if observation.ObservedAt.IsZero() { | ||
| observation.ObservedAt = time.Now() | ||
| } | ||
| if observation.AttemptLatencyMS < 0 { | ||
| observation.AttemptLatencyMS = 0 | ||
| } | ||
|
|
||
| ctx, cancel := context.WithTimeout(context.Background(), channelHealthObservationTimeout) | ||
| defer cancel() | ||
| key := channelHealthRuntimeKey(observation.ChannelID) | ||
| fields := map[string]interface{}{ | ||
| "channel_id": strconv.Itoa(observation.ChannelID), | ||
| "last_event": observation.Event, | ||
| "last_observed_at": observation.ObservedAt.UTC().Format(time.RFC3339Nano), | ||
| "last_attempt_latency": strconv.FormatInt(observation.AttemptLatencyMS, 10), | ||
| "last_stream": strconv.FormatBool(observation.Stream), | ||
| "last_retry_index": strconv.Itoa(observation.RetryIndex), | ||
| } | ||
|
|
||
| pipe := RDB.TxPipeline() | ||
| pipe.HIncrBy(ctx, key, "observations", 1) | ||
| if isChannelHealthTimeoutEvent(observation.Event) { | ||
| pipe.HIncrBy(ctx, key, "timeouts", 1) | ||
| } | ||
| pipe.HSet(ctx, key, fields) | ||
| pipe.Expire(ctx, key, channelHealthRuntimeTTL) | ||
| _, err := pipe.Exec(ctx) | ||
| if err != nil && !errors.Is(err, redis.Nil) { | ||
| return fmt.Errorf("record channel health observation: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/alicebob/miniredis/v2" | ||
| "github.com/go-redis/redis/v8" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestRecordChannelHealthObservationStoresBoundedRuntimeProjection(t *testing.T) { | ||
| server := miniredis.RunT(t) | ||
| oldRDB, oldEnabled := RDB, RedisEnabled | ||
| RDB = redis.NewClient(&redis.Options{Addr: server.Addr()}) | ||
| RedisEnabled = true | ||
| t.Cleanup(func() { | ||
| _ = RDB.Close() | ||
| RDB, RedisEnabled = oldRDB, oldEnabled | ||
| }) | ||
|
|
||
| now := time.Date(2026, 9, 30, 12, 0, 0, 0, time.UTC) | ||
| require.NoError(t, RecordChannelHealthObservation(ChannelHealthObservation{ | ||
| ChannelID: 936, | ||
| Event: "first_response", | ||
| ObservedAt: now, | ||
| AttemptLatencyMS: 1200, | ||
| Stream: true, | ||
| RetryIndex: 1, | ||
| })) | ||
| require.NoError(t, RecordChannelHealthObservation(ChannelHealthObservation{ | ||
| ChannelID: 936, | ||
| Event: "stream_idle_timeout", | ||
| ObservedAt: now.Add(time.Second), | ||
| AttemptLatencyMS: 2200, | ||
| Stream: true, | ||
| RetryIndex: 1, | ||
| })) | ||
|
|
||
| fields, err := RDB.HGetAll(context.Background(), channelHealthRuntimeKey(936)).Result() | ||
| require.NoError(t, err) | ||
| require.Equal(t, "2", fields["observations"]) | ||
| require.Equal(t, "1", fields["timeouts"]) | ||
| require.Equal(t, "stream_idle_timeout", fields["last_event"]) | ||
| require.Equal(t, "2200", fields["last_attempt_latency"]) | ||
| require.Equal(t, "true", fields["last_stream"]) | ||
| require.Equal(t, "1", fields["last_retry_index"]) | ||
| require.Equal(t, "936", fields["channel_id"]) | ||
| } | ||
|
|
||
| func TestRecordChannelHealthObservationCountsCanonicalTimeoutKinds(t *testing.T) { | ||
| server := miniredis.RunT(t) | ||
| oldRDB, oldEnabled := RDB, RedisEnabled | ||
| RDB = redis.NewClient(&redis.Options{Addr: server.Addr()}) | ||
| RedisEnabled = true | ||
| t.Cleanup(func() { | ||
| _ = RDB.Close() | ||
| RDB, RedisEnabled = oldRDB, oldEnabled | ||
| }) | ||
|
|
||
| for _, event := range []string{"streaming_first_result_timeout", "non_streaming_response_timeout"} { | ||
| require.NoError(t, RecordChannelHealthObservation(ChannelHealthObservation{ | ||
| ChannelID: 936, | ||
| Event: event, | ||
| })) | ||
| } | ||
| fields, err := RDB.HGetAll(context.Background(), channelHealthRuntimeKey(936)).Result() | ||
| require.NoError(t, err) | ||
| require.Equal(t, "2", fields["timeouts"]) | ||
| } | ||
|
|
||
| func TestRecordChannelHealthObservationSkipsRedisWhenDisabled(t *testing.T) { | ||
| oldRDB, oldEnabled := RDB, RedisEnabled | ||
| RDB = nil | ||
| RedisEnabled = false | ||
| t.Cleanup(func() { | ||
| RDB, RedisEnabled = oldRDB, oldEnabled | ||
| }) | ||
|
|
||
| require.NoError(t, RecordChannelHealthObservation(ChannelHealthObservation{ | ||
| ChannelID: 1, | ||
| Event: "first_response", | ||
| })) | ||
| } | ||
|
|
||
| func TestEnqueueChannelHealthObservationSkipsDisabledPolicy(t *testing.T) { | ||
| server := miniredis.RunT(t) | ||
| oldRDB, oldEnabled := RDB, RedisEnabled | ||
| RDB = redis.NewClient(&redis.Options{Addr: server.Addr()}) | ||
| RedisEnabled = true | ||
| t.Cleanup(func() { | ||
| _ = RDB.Close() | ||
| RDB, RedisEnabled = oldRDB, oldEnabled | ||
| }) | ||
|
|
||
| EnqueueChannelHealthObservation(ChannelHealthObservation{ | ||
| ChannelID: 937, | ||
| Event: "first_response", | ||
| }) | ||
| time.Sleep(50 * time.Millisecond) | ||
| count, err := RDB.Exists(context.Background(), channelHealthRuntimeKey(937)).Result() | ||
| require.NoError(t, err) | ||
| require.Zero(t, count) | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.