chaos/chaos-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'origin/main' into nightly

Browse Source 
Resolve 4 conflicts:
- relay/compatible_handler.go: accept main's refactor (postConsumeQuota -> service.PostTextConsumeQuota)
- service/quota.go: accept main's PostClaudeConsumeQuota deletion, keep nightly's tiered billing in PostWssConsumeQuota and PostAudioConsumeQuota
- web/src/i18n/locales/{en,zh-CN}.json: merge both sets of translation keys

Post-merge integration:
- Add tiered billing (TryTieredSettle, InjectTieredBillingInfo) to PostTextConsumeQuota
- Update tool pricing calls to use nightly's generic GetToolPriceForModel/GetToolPrice API
This commit is contained in:
CaIon
2026-04-02 00:39:13 +08:00
parent d385d7abfe a706f00287
commit 35d0704640
122 changed files with 8307 additions and 3923 deletions
Download Patch File Download Diff File Expand all files Collapse all files
relay/common/override.go
+231 -3
View File
@@ -21,10 +21,23 @@ var negativeIndexRegexp = regexp.MustCompile(`\.(-\d+)`)
const (
paramOverrideContextRequestHeaders = "request_headers"
paramOverrideContextHeaderOverride = "header_override"
paramOverrideContextAuditRecorder = "__param_override_audit_recorder"
)
var errSourceHeaderNotFound = errors.New("source header does not exist")
var paramOverrideKeyAuditPaths = map[string]struct{}{
"model": {},
"original_model": {},
"upstream_model": {},
"service_tier": {},
"inference_geo": {},
}
type paramOverrideAuditRecorder struct {
lines []string
}
type ConditionOperation struct {
Path string `json:"path"` // JSON路径
Mode string `json:"mode"` // full, prefix, suffix, contains, gt, gte, lt, lte
@@ -118,6 +131,7 @@ func ApplyParamOverride(jsonData []byte, paramOverride map[string]interface{}, c
if len(paramOverride) == 0 {
return jsonData, nil
}
auditRecorder := getParamOverrideAuditRecorder(conditionContext)
// 尝试断言为操作格式
if operations, ok := tryParseOperations(paramOverride); ok {
@@ -125,7 +139,7 @@ func ApplyParamOverride(jsonData []byte, paramOverride map[string]interface{}, c
workingJSON := jsonData
var err error
if len(legacyOverride) > 0 {
workingJSON, err = applyOperationsLegacy(workingJSON, legacyOverride)
workingJSON, err = applyOperationsLegacy(workingJSON, legacyOverride, auditRecorder)
if err != nil {
return nil, err
}
@@ -137,7 +151,7 @@ func ApplyParamOverride(jsonData []byte, paramOverride map[string]interface{}, c
}
// 直接使用旧方法
return applyOperationsLegacy(jsonData, paramOverride)
return applyOperationsLegacy(jsonData, paramOverride, auditRecorder)
}
func buildLegacyParamOverride(paramOverride map[string]interface{}) map[string]interface{} {
@@ -161,14 +175,200 @@ func ApplyParamOverrideWithRelayInfo(jsonData []byte, info *RelayInfo) ([]byte,
}
overrideCtx := BuildParamOverrideContext(info)
var recorder *paramOverrideAuditRecorder
if shouldEnableParamOverrideAudit(paramOverride) {
recorder = &paramOverrideAuditRecorder{}
overrideCtx[paramOverrideContextAuditRecorder] = recorder
}
result, err := ApplyParamOverride(jsonData, paramOverride, overrideCtx)
if err != nil {
return nil, err
}
syncRuntimeHeaderOverrideFromContext(info, overrideCtx)
if info != nil {
if recorder != nil {
info.ParamOverrideAudit = recorder.lines
} else {
info.ParamOverrideAudit = nil
}
}
return result, nil
}
func shouldEnableParamOverrideAudit(paramOverride map[string]interface{}) bool {
if common.DebugEnabled {
return true
}
if len(paramOverride) == 0 {
return false
}
if operations, ok := tryParseOperations(paramOverride); ok {
for _, operation := range operations {
if shouldAuditParamPath(strings.TrimSpace(operation.Path)) ||
shouldAuditParamPath(strings.TrimSpace(operation.To)) {
return true
}
}
for key := range buildLegacyParamOverride(paramOverride) {
if shouldAuditParamPath(strings.TrimSpace(key)) {
return true
}
}
return false
}
for key := range paramOverride {
if shouldAuditParamPath(strings.TrimSpace(key)) {
return true
}
}
return false
}
func getParamOverrideAuditRecorder(context map[string]interface{}) *paramOverrideAuditRecorder {
if context == nil {
return nil
}
recorder, _ := context[paramOverrideContextAuditRecorder].(*paramOverrideAuditRecorder)
return recorder
}
func (r *paramOverrideAuditRecorder) recordOperation(mode, path, from, to string, value interface{}) {
if r == nil {
return
}
line := buildParamOverrideAuditLine(mode, path, from, to, value)
if line == "" {
return
}
if lo.Contains(r.lines, line) {
return
}
r.lines = append(r.lines, line)
}
func shouldAuditParamPath(path string) bool {
path = strings.TrimSpace(path)
if path == "" {
return false
}
if common.DebugEnabled {
return true
}
_, ok := paramOverrideKeyAuditPaths[path]
return ok
}
func shouldAuditOperation(mode, path, from, to string) bool {
if common.DebugEnabled {
return true
}
for _, candidate := range []string{path, to} {
if shouldAuditParamPath(candidate) {
return true
}
}
return false
}
func formatParamOverrideAuditValue(value interface{}) string {
switch typed := value.(type) {
case nil:
return "<empty>"
case string:
return typed
default:
return common.GetJsonString(typed)
}
}
func buildParamOverrideAuditLine(mode, path, from, to string, value interface{}) string {
mode = strings.TrimSpace(mode)
path = strings.TrimSpace(path)
from = strings.TrimSpace(from)
to = strings.TrimSpace(to)
if !shouldAuditOperation(mode, path, from, to) {
return ""
}
switch mode {
case "set":
if path == "" {
return ""
}
return fmt.Sprintf("set %s = %s", path, formatParamOverrideAuditValue(value))
case "delete":
if path == "" {
return ""
}
return fmt.Sprintf("delete %s", path)
case "copy":
if from == "" || to == "" {
return ""
}
return fmt.Sprintf("copy %s -> %s", from, to)
case "move":
if from == "" || to == "" {
return ""
}
return fmt.Sprintf("move %s -> %s", from, to)
case "prepend":
if path == "" {
return ""
}
return fmt.Sprintf("prepend %s with %s", path, formatParamOverrideAuditValue(value))
case "append":
if path == "" {
return ""
}
return fmt.Sprintf("append %s with %s", path, formatParamOverrideAuditValue(value))
case "trim_prefix", "trim_suffix", "ensure_prefix", "ensure_suffix":
if path == "" {
return ""
}
return fmt.Sprintf("%s %s with %s", mode, path, formatParamOverrideAuditValue(value))
case "trim_space", "to_lower", "to_upper":
if path == "" {
return ""
}
return fmt.Sprintf("%s %s", mode, path)
case "replace", "regex_replace":
if path == "" {
return ""
}
return fmt.Sprintf("%s %s from %s to %s", mode, path, from, to)
case "set_header":
if path == "" {
return ""
}
return fmt.Sprintf("set_header %s = %s", path, formatParamOverrideAuditValue(value))
case "delete_header":
if path == "" {
return ""
}
return fmt.Sprintf("delete_header %s", path)
case "copy_header", "move_header":
if from == "" || to == "" {
return ""
}
return fmt.Sprintf("%s %s -> %s", mode, from, to)
case "pass_headers":
return fmt.Sprintf("pass_headers %s", formatParamOverrideAuditValue(value))
case "sync_fields":
if from == "" || to == "" {
return ""
}
return fmt.Sprintf("sync_fields %s -> %s", from, to)
case "return_error":
return fmt.Sprintf("return_error %s", formatParamOverrideAuditValue(value))
default:
if path == "" {
return mode
}
return fmt.Sprintf("%s %s", mode, path)
}
}
func getParamOverrideMap(info *RelayInfo) map[string]interface{} {
if info == nil || info.ChannelMeta == nil {
return nil
@@ -455,7 +655,7 @@ func compareNumeric(jsonValue, targetValue gjson.Result, operator string) (bool,
}
// applyOperationsLegacy 原参数覆盖方法
func applyOperationsLegacy(jsonData []byte, paramOverride map[string]interface{}) ([]byte, error) {
func applyOperationsLegacy(jsonData []byte, paramOverride map[string]interface{}, auditRecorder *paramOverrideAuditRecorder) ([]byte, error) {
reqMap := make(map[string]interface{})
err := common.Unmarshal(jsonData, &reqMap)
if err != nil {
@@ -464,6 +664,7 @@ func applyOperationsLegacy(jsonData []byte, paramOverride map[string]interface{}
for key, value := range paramOverride {
reqMap[key] = value
auditRecorder.recordOperation("set", key, "", "", value)
}
return common.Marshal(reqMap)
@@ -471,6 +672,7 @@ func applyOperationsLegacy(jsonData []byte, paramOverride map[string]interface{}
func applyOperations(jsonStr string, operations []ParamOperation, conditionContext map[string]interface{}) (string, error) {
context := ensureContextMap(conditionContext)
auditRecorder := getParamOverrideAuditRecorder(context)
contextJSON, err := marshalContextJSON(context)
if err != nil {
return "", fmt.Errorf("failed to marshal condition context: %v", err)
@@ -506,6 +708,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("delete", path, "", "", nil)
}
case "set":
for _, path := range opPaths {
@@ -516,11 +719,15 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("set", path, "", "", op.Value)
}
case "move":
opFrom := processNegativeIndex(result, op.From)
opTo := processNegativeIndex(result, op.To)
result, err = moveValue(result, opFrom, opTo)
if err == nil {
auditRecorder.recordOperation("move", "", opFrom, opTo, nil)
}
case "copy":
if op.From == "" || op.To == "" {
return "", fmt.Errorf("copy from/to is required")
@@ -528,12 +735,16 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
opFrom := processNegativeIndex(result, op.From)
opTo := processNegativeIndex(result, op.To)
result, err = copyValue(result, opFrom, opTo)
if err == nil {
auditRecorder.recordOperation("copy", "", opFrom, opTo, nil)
}
case "prepend":
for _, path := range opPaths {
result, err = modifyValue(result, path, op.Value, op.KeepOrigin, true)
if err != nil {
break
}
auditRecorder.recordOperation("prepend", path, "", "", op.Value)
}
case "append":
for _, path := range opPaths {
@@ -541,6 +752,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("append", path, "", "", op.Value)
}
case "trim_prefix":
for _, path := range opPaths {
@@ -548,6 +760,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("trim_prefix", path, "", "", op.Value)
}
case "trim_suffix":
for _, path := range opPaths {
@@ -555,6 +768,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("trim_suffix", path, "", "", op.Value)
}
case "ensure_prefix":
for _, path := range opPaths {
@@ -562,6 +776,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("ensure_prefix", path, "", "", op.Value)
}
case "ensure_suffix":
for _, path := range opPaths {
@@ -569,6 +784,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("ensure_suffix", path, "", "", op.Value)
}
case "trim_space":
for _, path := range opPaths {
@@ -576,6 +792,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("trim_space", path, "", "", nil)
}
case "to_lower":
for _, path := range opPaths {
@@ -583,6 +800,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("to_lower", path, "", "", nil)
}
case "to_upper":
for _, path := range opPaths {
@@ -590,6 +808,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("to_upper", path, "", "", nil)
}
case "replace":
for _, path := range opPaths {
@@ -597,6 +816,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("replace", path, op.From, op.To, nil)
}
case "regex_replace":
for _, path := range opPaths {
@@ -604,8 +824,10 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
if err != nil {
break
}
auditRecorder.recordOperation("regex_replace", path, op.From, op.To, nil)
}
case "return_error":
auditRecorder.recordOperation("return_error", op.Path, "", "", op.Value)
returnErr, parseErr := parseParamOverrideReturnError(op.Value)
if parseErr != nil {
return "", parseErr
@@ -621,11 +843,13 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
case "set_header":
err = setHeaderOverrideInContext(context, op.Path, op.Value, op.KeepOrigin)
if err == nil {
auditRecorder.recordOperation("set_header", op.Path, "", "", op.Value)
contextJSON, err = marshalContextJSON(context)
}
case "delete_header":
err = deleteHeaderOverrideInContext(context, op.Path)
if err == nil {
auditRecorder.recordOperation("delete_header", op.Path, "", "", nil)
contextJSON, err = marshalContextJSON(context)
}
case "copy_header":
@@ -642,6 +866,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
err = nil
}
if err == nil {
auditRecorder.recordOperation("copy_header", "", sourceHeader, targetHeader, nil)
contextJSON, err = marshalContextJSON(context)
}
case "move_header":
@@ -658,6 +883,7 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
err = nil
}
if err == nil {
auditRecorder.recordOperation("move_header", "", sourceHeader, targetHeader, nil)
contextJSON, err = marshalContextJSON(context)
}
case "pass_headers":
@@ -675,11 +901,13 @@ func applyOperations(jsonStr string, operations []ParamOperation, conditionConte
}
}
if err == nil {
auditRecorder.recordOperation("pass_headers", "", "", "", headerNames)
contextJSON, err = marshalContextJSON(context)
}
case "sync_fields":
result, err = syncFieldsBetweenTargets(result, context, op.From, op.To)
if err == nil {
auditRecorder.recordOperation("sync_fields", "", op.From, op.To, nil)
contextJSON, err = marshalContextJSON(context)
}
default:
relay/common/override_test.go
+100
View File
@@ -6,6 +6,7 @@ import (
"reflect"
"testing"
common2 "github.com/QuantumNous/new-api/common"
"github.com/QuantumNous/new-api/types"
"github.com/QuantumNous/new-api/dto"
@@ -2066,6 +2067,105 @@ func TestRemoveDisabledFieldsAllowInferenceGeo(t *testing.T) {
assertJSONEqual(t, `{"inference_geo":"eu","store":true}`, string(out))
}
func TestApplyParamOverrideWithRelayInfoRecordsOperationAuditInDebugMode(t *testing.T) {
originalDebugEnabled := common2.DebugEnabled
common2.DebugEnabled = true
t.Cleanup(func() {
common2.DebugEnabled = originalDebugEnabled
})
info := &RelayInfo{
ChannelMeta: &ChannelMeta{
ParamOverride: map[string]interface{}{
"operations": []interface{}{
map[string]interface{}{
"mode": "copy",
"from": "metadata.target_model",
"to": "model",
},
map[string]interface{}{
"mode": "set",
"path": "service_tier",
"value": "flex",
},
map[string]interface{}{
"mode": "set",
"path": "temperature",
"value": 0.1,
},
},
},
},
}
out, err := ApplyParamOverrideWithRelayInfo([]byte(`{
"model":"gpt-4.1",
"temperature":0.7,
"metadata":{"target_model":"gpt-4.1-mini"}
}`), info)
if err != nil {
t.Fatalf("ApplyParamOverrideWithRelayInfo returned error: %v", err)
}
assertJSONEqual(t, `{
"model":"gpt-4.1-mini",
"temperature":0.1,
"service_tier":"flex",
"metadata":{"target_model":"gpt-4.1-mini"}
}`, string(out))
expected := []string{
"copy metadata.target_model -> model",
"set service_tier = flex",
"set temperature = 0.1",
}
if !reflect.DeepEqual(info.ParamOverrideAudit, expected) {
t.Fatalf("unexpected param override audit, got %#v", info.ParamOverrideAudit)
}
}
func TestApplyParamOverrideWithRelayInfoRecordsOnlyKeyOperationsWhenDebugDisabled(t *testing.T) {
originalDebugEnabled := common2.DebugEnabled
common2.DebugEnabled = false
t.Cleanup(func() {
common2.DebugEnabled = originalDebugEnabled
})
info := &RelayInfo{
ChannelMeta: &ChannelMeta{
ParamOverride: map[string]interface{}{
"operations": []interface{}{
map[string]interface{}{
"mode": "copy",
"from": "metadata.target_model",
"to": "model",
},
map[string]interface{}{
"mode": "set",
"path": "temperature",
"value": 0.1,
},
},
},
},
}
_, err := ApplyParamOverrideWithRelayInfo([]byte(`{
"model":"gpt-4.1",
"temperature":0.7,
"metadata":{"target_model":"gpt-4.1-mini"}
}`), info)
if err != nil {
t.Fatalf("ApplyParamOverrideWithRelayInfo returned error: %v", err)
}
expected := []string{
"copy metadata.target_model -> model",
}
if !reflect.DeepEqual(info.ParamOverrideAudit, expected) {
t.Fatalf("unexpected param override audit, got %#v", info.ParamOverrideAudit)
}
}
func assertJSONEqual(t *testing.T, want, got string) {
t.Helper()
relay/common/relay_info.go
+4 -6
View File
@@ -150,6 +150,7 @@ type RelayInfo struct {
LastError *types.NewAPIError
RuntimeHeadersOverride map[string]interface{}
UseRuntimeHeadersOverride bool
ParamOverrideAudit []string
PriceData types.PriceData
@@ -167,6 +168,8 @@ type RelayInfo struct {
// 若为空，调用 GetFinalRequestRelayFormat 会回退到 RequestConversionChain 的最后一项或 RelayFormat。
FinalRequestRelayFormat types.RelayFormat
StreamStatus *StreamStatus
ThinkingContentInfo
TokenCountMeta
*ClaudeConvertInfo
@@ -343,15 +346,10 @@ func GenRelayInfoClaude(c *gin.Context, request dto.Request) *RelayInfo {
info.ClaudeConvertInfo = &ClaudeConvertInfo{
LastMessagesType: LastMessageTypeNone,
}
info.IsClaudeBetaQuery = c.Query("beta") == "true" || isClaudeBetaForced(c)
info.IsClaudeBetaQuery = c.Query("beta") == "true"
return info
}
func isClaudeBetaForced(c *gin.Context) bool {
channelOtherSettings, ok := common.GetContextKeyType[dto.ChannelOtherSettings](c, constant.ContextKeyChannelOtherSetting)
return ok && channelOtherSettings.ClaudeBetaQuery
}
func GenRelayInfoRerank(c *gin.Context, request *dto.RerankRequest) *RelayInfo {
info := genBaseRelayInfo(c, request)
info.RelayMode = relayconstant.RelayModeRerank
relay/common/stream_status.go
+112
View File
@@ -0,0 +1,112 @@
package common
import (
"fmt"
"strings"
"sync"
"time"
)
type StreamEndReason string
const (
StreamEndReasonNone StreamEndReason = ""
StreamEndReasonDone StreamEndReason = "done"
StreamEndReasonTimeout StreamEndReason = "timeout"
StreamEndReasonClientGone StreamEndReason = "client_gone"
StreamEndReasonScannerErr StreamEndReason = "scanner_error"
StreamEndReasonHandlerStop StreamEndReason = "handler_stop"
StreamEndReasonEOF StreamEndReason = "eof"
StreamEndReasonPanic StreamEndReason = "panic"
StreamEndReasonPingFail StreamEndReason = "ping_fail"
)
const maxStreamErrorEntries = 20
type StreamErrorEntry struct {
Message string
Timestamp time.Time
}
type StreamStatus struct {
EndReason StreamEndReason
EndError error
endOnce sync.Once
mu sync.Mutex
Errors []StreamErrorEntry
ErrorCount int
}
func NewStreamStatus() *StreamStatus {
return &StreamStatus{}
}
func (s *StreamStatus) SetEndReason(reason StreamEndReason, err error) {
if s == nil {
return
}
s.endOnce.Do(func() {
s.EndReason = reason
s.EndError = err
})
}
func (s *StreamStatus) RecordError(msg string) {
if s == nil {
return
}
s.mu.Lock()
defer s.mu.Unlock()
s.ErrorCount++
if len(s.Errors) < maxStreamErrorEntries {
s.Errors = append(s.Errors, StreamErrorEntry{
Message: msg,
Timestamp: time.Now(),
})
}
}
func (s *StreamStatus) HasErrors() bool {
if s == nil {
return false
}
s.mu.Lock()
defer s.mu.Unlock()
return s.ErrorCount > 0
}
func (s *StreamStatus) TotalErrorCount() int {
if s == nil {
return 0
}
s.mu.Lock()
defer s.mu.Unlock()
return s.ErrorCount
}
func (s *StreamStatus) IsNormalEnd() bool {
if s == nil {
return true
}
return s.EndReason == StreamEndReasonDone ||
s.EndReason == StreamEndReasonEOF ||
s.EndReason == StreamEndReasonHandlerStop
}
func (s *StreamStatus) Summary() string {
if s == nil {
return "StreamStatus<nil>"
}
b := &strings.Builder{}
fmt.Fprintf(b, "reason=%s", s.EndReason)
if s.EndError != nil {
fmt.Fprintf(b, " end_error=%q", s.EndError.Error())
}
s.mu.Lock()
if s.ErrorCount > 0 {
fmt.Fprintf(b, " soft_errors=%d", s.ErrorCount)
}
s.mu.Unlock()
return b.String()
}
relay/common/stream_status_test.go
+182
View File
@@ -0,0 +1,182 @@
package common
import (
"fmt"
"sync"
"testing"
"github.com/stretchr/testify/assert"
)
func TestStreamStatus_SetEndReason_FirstWins(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
s.SetEndReason(StreamEndReasonDone, nil)
s.SetEndReason(StreamEndReasonTimeout, nil)
s.SetEndReason(StreamEndReasonClientGone, fmt.Errorf("context canceled"))
assert.Equal(t, StreamEndReasonDone, s.EndReason)
assert.Nil(t, s.EndError)
}
func TestStreamStatus_SetEndReason_WithError(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
expectedErr := fmt.Errorf("read: connection reset")
s.SetEndReason(StreamEndReasonScannerErr, expectedErr)
assert.Equal(t, StreamEndReasonScannerErr, s.EndReason)
assert.Equal(t, expectedErr, s.EndError)
}
func TestStreamStatus_SetEndReason_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
s.SetEndReason(StreamEndReasonDone, nil)
}
func TestStreamStatus_SetEndReason_Concurrent(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
reasons := []StreamEndReason{
StreamEndReasonDone,
StreamEndReasonTimeout,
StreamEndReasonClientGone,
StreamEndReasonScannerErr,
StreamEndReasonHandlerStop,
StreamEndReasonEOF,
StreamEndReasonPanic,
StreamEndReasonPingFail,
}
var wg sync.WaitGroup
for _, r := range reasons {
wg.Add(1)
go func(reason StreamEndReason) {
defer wg.Done()
s.SetEndReason(reason, nil)
}(r)
}
wg.Wait()
assert.NotEqual(t, StreamEndReasonNone, s.EndReason)
}
func TestStreamStatus_RecordError_Basic(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
s.RecordError("bad json")
s.RecordError("another bad json")
s.RecordError("client gone")
assert.True(t, s.HasErrors())
assert.Equal(t, 3, s.TotalErrorCount())
assert.Len(t, s.Errors, 3)
}
func TestStreamStatus_RecordError_CapAtMax(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
for i := 0; i < 30; i++ {
s.RecordError(fmt.Sprintf("error_%d", i))
}
assert.Equal(t, maxStreamErrorEntries, len(s.Errors))
assert.Equal(t, 30, s.TotalErrorCount())
}
func TestStreamStatus_RecordError_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
s.RecordError("should not panic")
}
func TestStreamStatus_RecordError_Concurrent(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
go func(idx int) {
defer wg.Done()
s.RecordError(fmt.Sprintf("error_%d", idx))
}(i)
}
wg.Wait()
assert.Equal(t, 100, s.TotalErrorCount())
assert.LessOrEqual(t, len(s.Errors), maxStreamErrorEntries)
}
func TestStreamStatus_HasErrors_Empty(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
assert.False(t, s.HasErrors())
assert.Equal(t, 0, s.TotalErrorCount())
}
func TestStreamStatus_HasErrors_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
assert.False(t, s.HasErrors())
assert.Equal(t, 0, s.TotalErrorCount())
}
func TestStreamStatus_IsNormalEnd(t *testing.T) {
t.Parallel()
tests := []struct {
reason StreamEndReason
normal bool
}{
{StreamEndReasonDone, true},
{StreamEndReasonEOF, true},
{StreamEndReasonHandlerStop, true},
{StreamEndReasonTimeout, false},
{StreamEndReasonClientGone, false},
{StreamEndReasonScannerErr, false},
{StreamEndReasonPanic, false},
{StreamEndReasonPingFail, false},
{StreamEndReasonNone, false},
}
for _, tt := range tests {
s := NewStreamStatus()
s.SetEndReason(tt.reason, nil)
assert.Equal(t, tt.normal, s.IsNormalEnd(), "reason=%s", tt.reason)
}
}
func TestStreamStatus_IsNormalEnd_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
assert.True(t, s.IsNormalEnd())
}
func TestStreamStatus_Summary(t *testing.T) {
t.Parallel()
s := NewStreamStatus()
s.SetEndReason(StreamEndReasonDone, nil)
summary := s.Summary()
assert.Contains(t, summary, "reason=done")
assert.NotContains(t, summary, "soft_errors")
s2 := NewStreamStatus()
s2.SetEndReason(StreamEndReasonTimeout, nil)
s2.RecordError("bad json")
s2.RecordError("write failed")
summary2 := s2.Summary()
assert.Contains(t, summary2, "reason=timeout")
assert.Contains(t, summary2, "soft_errors=2")
}
func TestStreamStatus_Summary_NilSafe(t *testing.T) {
t.Parallel()
var s *StreamStatus
assert.Equal(t, "StreamStatus<nil>", s.Summary())
}