diff --git a/internal/control/credential_observations.go b/internal/control/credential_observations.go index 084fa6a6..de5071e2 100644 --- a/internal/control/credential_observations.go +++ b/internal/control/credential_observations.go @@ -465,16 +465,23 @@ func observationWindowRemainingRatio(window ObservationQuotaWindow) (float64, bo } // credentialDailyUsageWindow 是账号卡上「近 24 小时成功/失败」的窗口长度。 -// 取 24 小时是因为 QueryCredentialWindowUsage 在不超过 24 小时时走精确请求日志， -// 再长就退化成小时聚合，边界会变成近似值。 const credentialDailyUsageWindow = 24 * time.Hour -func (s *Service) enrichCredentialDailyUsage( +func (s *Service) enrichCredentialActivities( ctx context.Context, - credentialID uint, - item *CredentialItemResponse, + items []CredentialItemResponse, ) { - if s == nil || s.credentialWindowUsage == nil || credentialID == 0 || item == nil { + if s == nil || s.credentialActivity == nil || len(items) == 0 { + return + } + credentialIDs := make([]uint, 0, len(items)) + for index := range items { + if items[index].ConnectionType == string(models.ConnectionTypeSubscription) && + items[index].CredentialID != 0 { + credentialIDs = append(credentialIDs, items[index].CredentialID) + } + } + if len(credentialIDs) == 0 { return } now := s.now().UTC() @@ -483,33 +490,60 @@ func (s *Service) enrichCredentialDailyUsage( if fromMS < 0 || toMS <= fromMS { return } - observed, err := s.credentialWindowUsage.QueryCredentialWindowUsage(ctx, requestlog.CredentialWindowUsageQuery{ - CredentialID: credentialID, - FromMS: fromMS, - ToMS: toMS, - Source: requestlog.CredentialWindowUsageSourceRequestLogs, - }) + activities, err := s.credentialActivity.QueryCredentialActivity( + ctx, + requestlog.CredentialActivityQuery{ + CredentialIDs: credentialIDs, + FromMS: fromMS, + ToMS: toMS, + }, + ) if err != nil { return } - item.DailyUsage = &CredentialDailyUsageResponse{ - WindowSeconds: int64(credentialDailyUsageWindow / time.Second), - SuccessCount: observed.SuccessCount, - FailureCount: observed.FailureCount, - DataComplete: observed.DataComplete, + for index := range items { + activity, exists := activities[items[index].CredentialID] + if !exists { + continue + } + items[index].LastUsedAtMS = activity.LastUsedAtMS + items[index].DailyUsage = &CredentialDailyUsageResponse{ + WindowSeconds: int64(credentialDailyUsageWindow / time.Second), + SuccessCount: activity.SuccessCount, + FailureCount: activity.FailureCount, + DataComplete: activity.DataComplete, + } } } +func (s *Service) enrichCredentialDailyUsage( + ctx context.Context, + credentialID uint, + item *CredentialItemResponse, +) { + if item == nil || credentialID == 0 { + return + } + item.CredentialID = credentialID + items := []CredentialItemResponse{*item} + s.enrichCredentialActivities(ctx, items) + *item = items[0] +} + func (s *Service) enrichCredentialObservationUsage( ctx context.Context, credentialID uint, response *CredentialObservationResponse, ) { if s == nil || s.credentialWindowUsage == nil || credentialID == 0 || - response == nil || response.Snapshot == nil { + response == nil || response.State != string(models.CredentialObservationFresh) || + response.Snapshot == nil || response.FreshUntilMS == nil { return } nowMS := s.now().UTC().UnixMilli() + if *response.FreshUntilMS <= nowMS { + return + } for index := range response.Snapshot.QuotaWindows { window := &response.Snapshot.QuotaWindows[index] if window.Scope != "account" || window.ResetAtMS == nil || @@ -528,15 +562,11 @@ func (s *Service) enrichCredentialObservationUsage( if fromMS >= nowMS { continue } - source := requestlog.CredentialWindowUsageSourceRequestLogs - if *window.WindowSeconds > int64((24*time.Hour)/time.Second) { - source = requestlog.CredentialWindowUsageSourceHourlyStats - } observed, err := s.credentialWindowUsage.QueryCredentialWindowUsage(ctx, requestlog.CredentialWindowUsageQuery{ CredentialID: credentialID, FromMS: fromMS, ToMS: nowMS, - Source: source, + Source: requestlog.CredentialWindowUsageSourceHourlyStats, }) if err != nil { continue diff --git a/internal/control/credential_observations_test.go b/internal/control/credential_observations_test.go index 7c37aae7..b56feceb 100644 --- a/internal/control/credential_observations_test.go +++ b/internal/control/credential_observations_test.go @@ -204,18 +204,20 @@ func TestRefreshCredentialObservationKeepsUsageWhenResetCreditDetailsFail(t *tes } } -func TestEnrichCredentialObservationUsageSelectsExactAndHourlySources(t *testing.T) { +func TestEnrichCredentialObservationUsageUsesHourlyStatsForEveryWindow(t *testing.T) { fixture := newServiceFixture(t) now := time.Date(2026, time.August, 14, 12, 0, 0, 0, time.UTC) fixture.service.now = func() time.Time { return now } shortReset := now.Add(2 * time.Hour).UnixMilli() longReset := now.Add(3 * 24 * time.Hour).UnixMilli() modelReset := now.Add(time.Hour).UnixMilli() + freshUntil := now.Add(time.Hour).UnixMilli() shortSeconds := int64((5 * time.Hour) / time.Second) longSeconds := int64((7 * 24 * time.Hour) / time.Second) modelSeconds := int64(time.Hour / time.Second) response := CredentialObservationResponse{ - State: string(models.CredentialObservationFresh), + State: string(models.CredentialObservationFresh), + FreshUntilMS: &freshUntil, Snapshot: &CredentialObservationSnapshot{QuotaWindows: []ObservationQuotaWindow{ {ID: "short", Scope: "account", ResetAtMS: &shortReset, WindowSeconds: &shortSeconds}, {ID: "long", Scope: "account", ResetAtMS: &longReset, WindowSeconds: &longSeconds}, @@ -237,7 +239,7 @@ func TestEnrichCredentialObservationUsageSelectsExactAndHourlySources(t *testing if len(reader.queries) != 2 { t.Fatalf("usage queries = %#v, want two account windows", reader.queries) } - if reader.queries[0].Source != requestlog.CredentialWindowUsageSourceRequestLogs || + if reader.queries[0].Source != requestlog.CredentialWindowUsageSourceHourlyStats || reader.queries[0].FromMS != shortReset-shortSeconds*1000 || reader.queries[0].ToMS != now.UnixMilli() { t.Fatalf("short window query = %#v", reader.queries[0]) @@ -257,6 +259,46 @@ func TestEnrichCredentialObservationUsageSelectsExactAndHourlySources(t *testing } } +func TestEnrichCredentialObservationUsageSkipsNonCurrentObservation(t *testing.T) { + now := time.Date(2026, time.August, 14, 12, 0, 0, 0, time.UTC) + future := now.Add(time.Hour).UnixMilli() + expired := now.UnixMilli() + resetAt := now.Add(2 * time.Hour).UnixMilli() + windowSeconds := int64((5 * time.Hour) / time.Second) + tests := []struct { + name string + state string + freshUntilMS *int64 + }{ + {name: "stale", state: string(models.CredentialObservationStale), freshUntilMS: &future}, + {name: "error", state: string(models.CredentialObservationError), freshUntilMS: &future}, + {name: "missing fresh until", state: string(models.CredentialObservationFresh)}, + {name: "expired", state: string(models.CredentialObservationFresh), freshUntilMS: &expired}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fixture := newServiceFixture(t) + fixture.service.now = func() time.Time { return now } + reader := &recordingCredentialWindowUsageReader{} + fixture.service.credentialWindowUsage = reader + response := CredentialObservationResponse{ + State: test.state, + FreshUntilMS: test.freshUntilMS, + Snapshot: &CredentialObservationSnapshot{QuotaWindows: []ObservationQuotaWindow{{ + ID: "account", Scope: "account", ResetAtMS: &resetAt, WindowSeconds: &windowSeconds, + }}}, + } + + fixture.service.enrichCredentialObservationUsage(t.Context(), 37, &response) + + if len(reader.queries) != 0 || response.Snapshot.QuotaWindows[0].ObservedUsage != nil { + t.Fatalf("non-current observation was enriched: queries=%#v response=%#v", reader.queries, response) + } + }) + } +} + func TestRefreshCredentialObservationPublishesAndClearsQuotaRoutingState(t *testing.T) { fixture, groupID, credentialID := newSubscriptionCredentialFixture(t) now := time.Date(2026, time.August, 14, 15, 0, 0, 0, time.UTC) @@ -429,19 +471,35 @@ func (reader *recordingCredentialWindowUsageReader) QueryCredentialWindowUsage( return result, nil } -func TestEnrichCredentialDailyUsageQueriesExactDayWindow(t *testing.T) { +type recordingCredentialActivityReader struct { + queries []requestlog.CredentialActivityQuery + result map[uint]requestlog.CredentialActivity + err error +} + +func (reader *recordingCredentialActivityReader) QueryCredentialActivity( + _ context.Context, + query requestlog.CredentialActivityQuery, +) (map[uint]requestlog.CredentialActivity, error) { + reader.queries = append(reader.queries, query) + return reader.result, reader.err +} + +func TestEnrichCredentialDailyUsageQueriesAttemptActivity(t *testing.T) { fixture, _, _ := newSubscriptionCredentialFixture(t) now := time.UnixMilli(1_800_000_000_000) fixture.service.now = func() time.Time { return now } - reader := &recordingCredentialWindowUsageReader{} - reader.result = requestlog.CredentialWindowUsage{ - UsageAggregate: requestlog.UsageAggregate{ - RequestCount: 18, SuccessCount: 15, FailureCount: 3, + lastUsedAtMS := now.Add(-time.Minute).UnixMilli() + reader := &recordingCredentialActivityReader{result: map[uint]requestlog.CredentialActivity{ + 37: { + CredentialID: 37, LastUsedAtMS: &lastUsedAtMS, + SuccessCount: 15, FailureCount: 3, DataComplete: true, }, - DataComplete: true, + }} + fixture.service.credentialActivity = reader + item := CredentialItemResponse{ + CredentialID: 37, ConnectionType: string(models.ConnectionTypeSubscription), } - fixture.service.credentialWindowUsage = reader - item := CredentialItemResponse{CredentialID: 37} fixture.service.enrichCredentialDailyUsage(t.Context(), 37, &item) @@ -449,25 +507,26 @@ func TestEnrichCredentialDailyUsageQueriesExactDayWindow(t *testing.T) { t.Fatalf("usage queries = %#v, want exactly one day window", reader.queries) } query := reader.queries[0] - // 24 小时窗口必须走精确请求日志，走小时聚合会把窗口边界变成近似值。 - if query.Source != requestlog.CredentialWindowUsageSourceRequestLogs || - query.CredentialID != 37 || + if !reflect.DeepEqual(query.CredentialIDs, []uint{37}) || query.ToMS != now.UnixMilli() || query.FromMS != now.Add(-24*time.Hour).UnixMilli() { t.Fatalf("day window query = %#v", query) } if item.DailyUsage == nil || item.DailyUsage.WindowSeconds != 86_400 || item.DailyUsage.SuccessCount != 15 || item.DailyUsage.FailureCount != 3 || - !item.DailyUsage.DataComplete { - t.Fatalf("daily usage = %#v", item.DailyUsage) + !item.DailyUsage.DataComplete || item.LastUsedAtMS == nil || + *item.LastUsedAtMS != lastUsedAtMS { + t.Fatalf("credential activity = %#v / %#v", item.LastUsedAtMS, item.DailyUsage) } } func TestEnrichCredentialDailyUsageLeavesItemUntouchedOnQueryFailure(t *testing.T) { fixture, _, _ := newSubscriptionCredentialFixture(t) fixture.service.now = func() time.Time { return time.UnixMilli(1_800_000_000_000) } - fixture.service.credentialWindowUsage = failingCredentialWindowUsageReader{} - item := CredentialItemResponse{CredentialID: 37} + fixture.service.credentialActivity = &recordingCredentialActivityReader{err: errors.New("activity unavailable")} + item := CredentialItemResponse{ + CredentialID: 37, ConnectionType: string(models.ConnectionTypeSubscription), + } fixture.service.enrichCredentialDailyUsage(t.Context(), 37, &item) @@ -476,13 +535,32 @@ func TestEnrichCredentialDailyUsageLeavesItemUntouchedOnQueryFailure(t *testing. } } -type failingCredentialWindowUsageReader struct{} +func TestEnrichCredentialActivitiesBatchesSubscriptionItems(t *testing.T) { + fixture := newServiceFixture(t) + now := time.UnixMilli(1_800_000_000_000) + fixture.service.now = func() time.Time { return now } + reader := &recordingCredentialActivityReader{result: map[uint]requestlog.CredentialActivity{ + 37: {CredentialID: 37, SuccessCount: 4, DataComplete: true}, + 39: {CredentialID: 39, FailureCount: 2, DataComplete: true}, + }} + fixture.service.credentialActivity = reader + items := []CredentialItemResponse{ + {CredentialID: 37, ConnectionType: string(models.ConnectionTypeSubscription)}, + {CredentialID: 38, ConnectionType: string(models.ConnectionTypeAPIKey)}, + {CredentialID: 39, ConnectionType: string(models.ConnectionTypeSubscription)}, + } -func (failingCredentialWindowUsageReader) QueryCredentialWindowUsage( - _ context.Context, - _ requestlog.CredentialWindowUsageQuery, -) (requestlog.CredentialWindowUsage, error) { - return requestlog.CredentialWindowUsage{}, errors.New("request logs unavailable") + fixture.service.enrichCredentialActivities(t.Context(), items) + + if len(reader.queries) != 1 || + !reflect.DeepEqual(reader.queries[0].CredentialIDs, []uint{37, 39}) { + t.Fatalf("activity queries = %#v, want one subscription batch", reader.queries) + } + if items[0].DailyUsage == nil || items[0].DailyUsage.SuccessCount != 4 || + items[1].DailyUsage != nil || items[2].DailyUsage == nil || + items[2].DailyUsage.FailureCount != 2 { + t.Fatalf("enriched items = %#v", items) + } } func TestRefreshCredentialObservationPersistsNormalizationFailure(t *testing.T) { diff --git a/internal/control/credentials.go b/internal/control/credentials.go index 718ab9dc..83c22a09 100644 --- a/internal/control/credentials.go +++ b/internal/control/credentials.go @@ -83,18 +83,19 @@ type CredentialItemResponse struct { LastFailureCategory string `json:"last_failure_category"` LastStatusCode *int `json:"last_status_code"` CooldownUntilMS *int64 `json:"cooldown_until_ms"` + LastUsedAtMS *int64 `json:"last_used_at_ms,omitempty"` DailyUsage *CredentialDailyUsageResponse `json:"daily_usage,omitempty"` Recovery CredentialRecoveryResponse `json:"recovery"` } -// CredentialDailyUsageResponse 汇报固定 24 小时窗口内的请求结果分布。 +// CredentialDailyUsageResponse 汇报固定 24 小时窗口内的上游尝试结果分布。 // recent_success_count / recent_failure_count 来自 health 的 5 分钟内存窗口， -// 是调度判定用的，重启即清零；这里的计数来自请求日志，用于给人看。 +// 是调度判定用的，重启即清零；这里的计数来自账号小时聚合，用于给人看。 type CredentialDailyUsageResponse struct { WindowSeconds int64 `json:"window_seconds"` SuccessCount int64 `json:"success_count"` FailureCount int64 `json:"failure_count"` - // DataComplete 为 false 表示请求日志留存期短于该窗口，计数偏低。 + // DataComplete 为 false 表示统计数据未覆盖完整窗口，计数偏低。 DataComplete bool `json:"data_complete"` } @@ -426,12 +427,8 @@ func (s *Service) mapCredentialCollection( }) total := len(filtered) items := credentialCollectionPage(filtered, query.Page, query.PageSize) - for index := range items { - s.enrichCredentialObservationUsage(ctx, items[index].CredentialID, items[index].Observation) - if items[index].ConnectionType == string(models.ConnectionTypeSubscription) { - s.enrichCredentialDailyUsage(ctx, items[index].CredentialID, &items[index]) - } - } + // 配额窗口的本地用量只在账号详情被展开时读取，避免列表按卡片聚合。 + s.enrichCredentialActivities(ctx, items) return CredentialCollectionResponse{ ObservedAtMS: observedAtMS, StatsWindowSeconds: credentialCollectionStatsWindow, Summary: summary, Items: items, diff --git a/internal/control/service.go b/internal/control/service.go index c6774492..12bfbf99 100644 --- a/internal/control/service.go +++ b/internal/control/service.go @@ -50,6 +50,7 @@ type Service struct { requestLogs RequestLogReader usageStats UsageStatReader credentialWindowUsage credentialWindowUsageReader + credentialActivity credentialActivityReader homeStatistics HomeStatisticsReader stats *health.StatsStore mutations credentialMutationCoordinator @@ -93,6 +94,13 @@ type credentialWindowUsageReader interface { ) (requestlog.CredentialWindowUsage, error) } +type credentialActivityReader interface { + QueryCredentialActivity( + context.Context, + requestlog.CredentialActivityQuery, + ) (map[uint]requestlog.CredentialActivity, error) +} + type credentialMultiMutationCoordinator interface { DoMany([]uint, func()) } @@ -193,6 +201,11 @@ func NewService( } else if reader, ok := requestLogs.(credentialWindowUsageReader); ok { service.credentialWindowUsage = reader } + if reader, ok := usageStats.(credentialActivityReader); ok { + service.credentialActivity = reader + } else if reader, ok := requestLogs.(credentialActivityReader); ok { + service.credentialActivity = reader + } if provider, ok := executor.(channelDefaultBaseURLProvider); ok { service.channelDefaultBaseURLs = provider } diff --git a/internal/gateway/request_log.go b/internal/gateway/request_log.go index a9f9a10a..733f5bd7 100644 --- a/internal/gateway/request_log.go +++ b/internal/gateway/request_log.go @@ -313,6 +313,7 @@ func (recorder *requestRecorder) appendAttempt( } attempt := telemetry.Attempt{ Sequence: len(recorder.attempts) + 1, + CompletedAt: completedAt, GroupID: selection.GroupID, GroupName: selection.Group.Name, ChannelID: selection.ChannelID, diff --git a/internal/gateway/request_log_test.go b/internal/gateway/request_log_test.go index cee26748..b92649d0 100644 --- a/internal/gateway/request_log_test.go +++ b/internal/gateway/request_log_test.go @@ -45,6 +45,33 @@ func requestLogSelection(groupID, credentialID uint, name string) scheduler.Sele } } +func TestRequestRecorderCapturesAttemptCompletionTime(t *testing.T) { + completedAt := time.Date(2026, time.August, 15, 14, 35, 27, 123_000_000, time.UTC) + recorder := newRequestRecorder( + &recordingRequestLogSink{}, + "00000000-0000-4000-8000-000000000001", + completedAt.Add(-time.Second), + 1, + protocol.OpenAICompletions, + func() time.Time { return completedAt }, + ) + + recorder.appendAttempt( + requestLogSelection(11, 21, "first"), + UpstreamResult{StatusCode: http.StatusOK}, + telemetry.FailureCategoryOK, + telemetry.ActionTerminate, + "", + "", + completedAt.Add(-250*time.Millisecond), + completedAt, + ) + + if len(recorder.attempts) != 1 || !recorder.attempts[0].CompletedAt.Equal(completedAt) { + t.Fatalf("attempt completion = %#v, want %s", recorder.attempts, completedAt) + } +} + func TestRequestRecorderCapturesFirstResponseAfterDownstreamCommit(t *testing.T) { startedAt := time.Date(2026, time.August, 5, 8, 0, 0, 0, time.UTC) sink := &recordingRequestLogSink{} diff --git a/internal/requestlog/credential_activity.go b/internal/requestlog/credential_activity.go new file mode 100644 index 00000000..2d57b901 --- /dev/null +++ b/internal/requestlog/credential_activity.go @@ -0,0 +1,249 @@ +package requestlog + +import ( + "context" + "fmt" + "strings" + + "gorm.io/gorm" + + "gpt-load/internal/platform/epochms" + "gpt-load/internal/storage/dbtx" + "gpt-load/internal/storage/models" + "gpt-load/internal/telemetry" + "gpt-load/internal/usage" +) + +const ( + maxCredentialActivityIDs = 100 + credentialActivityMaxWindowMS = 24 * epochms.MillisecondsPerHour +) + +type CredentialActivityQuery struct { + CredentialIDs []uint + FromMS int64 + ToMS int64 +} + +type CredentialActivity struct { + CredentialID uint + LastUsedAtMS *int64 + SuccessCount int64 + FailureCount int64 + DataComplete bool +} + +// QueryCredentialActivity returns the account-card activity for the requested +// credentials without issuing one query per account. Full hours come from the +// durable hourly aggregate; only the oldest partial hour reads attempt rows. +func (service *Service) QueryCredentialActivity( + ctx context.Context, + input CredentialActivityQuery, +) (map[uint]CredentialActivity, error) { + if service == nil || service.db == nil { + return nil, fmt.Errorf("query credential activity: database is nil") + } + credentialIDs, err := validateCredentialActivityQuery(input) + if err != nil { + return nil, err + } + if ctx == nil { + ctx = context.Background() + } + + hourlyFromMS, err := alignHourUp(input.FromMS) + if err != nil { + return nil, fmt.Errorf("query credential activity: align window start: %w", err) + } + hourlyToMS, err := alignHourUp(input.ToMS) + if err != nil { + return nil, fmt.Errorf("query credential activity: align window end: %w", err) + } + boundaryToMS := hourlyFromMS + if boundaryToMS > input.ToMS { + boundaryToMS = input.ToMS + } + hasHourlyWindow := hourlyFromMS < hourlyToMS + hasBoundaryWindow := input.FromMS < boundaryToMS + dataComplete := (!hasHourlyWindow || service.hourlyWindowRetained(hourlyFromMS)) && + (!hasBoundaryWindow || service.requestLogWindowRetained(input.FromMS)) + + result := make(map[uint]CredentialActivity, len(credentialIDs)) + for _, credentialID := range credentialIDs { + result[credentialID] = CredentialActivity{ + CredentialID: credentialID, + DataComplete: dataComplete, + } + } + + err = dbtx.Run(ctx, service.db, dbtx.Options{ + Mode: dbtx.ReadSnapshot, + CleanupTimeout: usageRollbackTimeout, + Operation: "credential activity read transaction", + }, func(connection *gorm.DB) error { + if err := queryCredentialLatestActivity( + connection, credentialIDs, input.ToMS, result, + ); err != nil { + return err + } + if hasHourlyWindow { + if err := queryCredentialHourlyActivity( + connection, credentialIDs, hourlyFromMS, hourlyToMS, result, + ); err != nil { + return err + } + } + if hasBoundaryWindow { + if err := queryCredentialBoundaryActivity( + connection, credentialIDs, input.FromMS, boundaryToMS, result, + ); err != nil { + return err + } + } + return nil + }) + if err != nil { + return nil, fmt.Errorf("query credential activity: %w", err) + } + return result, nil +} + +func validateCredentialActivityQuery(input CredentialActivityQuery) ([]uint, error) { + if len(input.CredentialIDs) == 0 || len(input.CredentialIDs) > maxCredentialActivityIDs || + input.FromMS < 0 || input.ToMS <= input.FromMS || + input.ToMS-input.FromMS > credentialActivityMaxWindowMS { + return nil, fmt.Errorf("query credential activity: invalid scope") + } + credentialIDs := make([]uint, 0, len(input.CredentialIDs)) + seen := make(map[uint]struct{}, len(input.CredentialIDs)) + for _, credentialID := range input.CredentialIDs { + if credentialID == 0 { + return nil, fmt.Errorf("query credential activity: invalid scope") + } + if _, exists := seen[credentialID]; exists { + continue + } + seen[credentialID] = struct{}{} + credentialIDs = append(credentialIDs, credentialID) + } + return credentialIDs, nil +} + +type credentialLatestActivityRow struct { + CredentialID uint `gorm:"column:credential_id"` + LastUsedAtMS *int64 `gorm:"column:last_used_at_ms"` +} + +func queryCredentialLatestActivity( + db *gorm.DB, + credentialIDs []uint, + toMS int64, + result map[uint]CredentialActivity, +) error { + var query strings.Builder + args := make([]any, 0, len(credentialIDs)*3) + for index, credentialID := range credentialIDs { + if index > 0 { + query.WriteString(" UNION ALL ") + } + query.WriteString(`SELECT ? AS credential_id, ( + SELECT completed_at_ms + FROM request_log_attempts + WHERE credential_id = ? AND completed_at_ms < ? + ORDER BY completed_at_ms DESC + LIMIT 1 + ) AS last_used_at_ms`) + args = append(args, credentialID, credentialID, toMS) + } + var rows []credentialLatestActivityRow + if err := db.Raw(query.String(), args...).Scan(&rows).Error; err != nil { + return fmt.Errorf("query latest credential attempts: %w", err) + } + for _, row := range rows { + activity, exists := result[row.CredentialID] + if !exists { + continue + } + activity.LastUsedAtMS = row.LastUsedAtMS + result[row.CredentialID] = activity + } + return nil +} + +type credentialActivityCountRow struct { + CredentialID uint `gorm:"column:credential_id"` + SuccessCount int64 `gorm:"column:success_count"` + FailureCount int64 `gorm:"column:failure_count"` +} + +func queryCredentialHourlyActivity( + db *gorm.DB, + credentialIDs []uint, + fromMS int64, + toMS int64, + result map[uint]CredentialActivity, +) error { + var rows []credentialActivityCountRow + if err := db.Model(&models.CredentialAttemptStat{}). + Select("credential_id, COALESCE(SUM(success_count), 0) AS success_count, "+ + "COALESCE(SUM(failure_count), 0) AS failure_count"). + Where("credential_id IN ?", credentialIDs). + Where("bucket_start_ms >= ? AND bucket_start_ms < ?", fromMS, toMS). + Group("credential_id"). + Find(&rows).Error; err != nil { + return fmt.Errorf("query hourly credential attempts: %w", err) + } + return mergeCredentialActivityCounts(result, rows) +} + +func queryCredentialBoundaryActivity( + db *gorm.DB, + credentialIDs []uint, + fromMS int64, + toMS int64, + result map[uint]CredentialActivity, +) error { + var rows []credentialActivityCountRow + if err := db.Model(&models.RequestLogAttempt{}). + Select("credential_id, "+ + "COALESCE(SUM(CASE WHEN failure_category = ? THEN 1 ELSE 0 END), 0) AS success_count, "+ + "COALESCE(SUM(CASE WHEN failure_category <> ? AND failure_category <> ? THEN 1 ELSE 0 END), 0) AS failure_count", + string(telemetry.FailureCategoryOK), + string(telemetry.FailureCategoryOK), + string(telemetry.FailureCategoryDownstreamCancel), + ). + Where("credential_id IN ?", credentialIDs). + Where("completed_at_ms >= ? AND completed_at_ms < ?", fromMS, toMS). + Group("credential_id"). + Find(&rows).Error; err != nil { + return fmt.Errorf("query credential attempt boundary: %w", err) + } + return mergeCredentialActivityCounts(result, rows) +} + +func mergeCredentialActivityCounts( + result map[uint]CredentialActivity, + rows []credentialActivityCountRow, +) error { + for _, row := range rows { + activity, exists := result[row.CredentialID] + if !exists { + continue + } + if row.SuccessCount < 0 || row.FailureCount < 0 { + return fmt.Errorf("validate credential activity: negative count") + } + successCount, ok := usage.CheckedAdd(activity.SuccessCount, row.SuccessCount) + if !ok { + return fmt.Errorf("aggregate credential activity success count: checked addition failed") + } + failureCount, ok := usage.CheckedAdd(activity.FailureCount, row.FailureCount) + if !ok { + return fmt.Errorf("aggregate credential activity failure count: checked addition failed") + } + activity.SuccessCount = successCount + activity.FailureCount = failureCount + result[row.CredentialID] = activity + } + return nil +} diff --git a/internal/requestlog/credential_activity_test.go b/internal/requestlog/credential_activity_test.go new file mode 100644 index 00000000..f3f55a2f --- /dev/null +++ b/internal/requestlog/credential_activity_test.go @@ -0,0 +1,169 @@ +package requestlog + +import ( + "context" + "fmt" + "testing" + "time" + + "gorm.io/gorm" + + "gpt-load/internal/storage/models" + "gpt-load/internal/telemetry" +) + +func TestQueryCredentialActivityCombinesHourlyStatsAndOldestPartialHour(t *testing.T) { + db := openRequestLogQueryDB(t) + service := newRequestLogTestService(db) + now := time.Date(2026, time.August, 15, 13, 45, 0, 0, time.UTC) + service.now = func() time.Time { return now } + from := now.Add(-24 * time.Hour) + + createCredentialAttemptStats(t, db, + models.CredentialAttemptStat{ + CredentialID: 11, BucketStartMS: from.Truncate(time.Hour).UnixMilli(), + SuccessCount: 100, FailureCount: 100, + }, + models.CredentialAttemptStat{ + CredentialID: 11, BucketStartMS: from.Truncate(time.Hour).Add(time.Hour).UnixMilli(), + SuccessCount: 2, FailureCount: 1, + }, + models.CredentialAttemptStat{ + CredentialID: 11, BucketStartMS: now.Truncate(time.Hour).UnixMilli(), + SuccessCount: 3, FailureCount: 4, + }, + models.CredentialAttemptStat{ + CredentialID: 12, BucketStartMS: now.Truncate(time.Hour).UnixMilli(), + SuccessCount: 7, FailureCount: 8, + }, + models.CredentialAttemptStat{ + CredentialID: 13, BucketStartMS: now.Truncate(time.Hour).UnixMilli(), + SuccessCount: 200, FailureCount: 300, + }, + ) + + createCredentialActivityAttempt(t, db, 1, 11, from.Add(5*time.Minute), telemetry.FailureCategoryOK) + createCredentialActivityAttempt(t, db, 2, 11, from.Add(6*time.Minute), telemetry.FailureCategoryRateLimited) + createCredentialActivityAttempt(t, db, 3, 11, from.Add(7*time.Minute), telemetry.FailureCategoryDownstreamCancel) + createCredentialActivityAttempt(t, db, 4, 11, from.Add(-time.Minute), telemetry.FailureCategoryAmbiguous) + latest11 := now.Add(-time.Minute) + createCredentialActivityAttempt(t, db, 5, 11, latest11, telemetry.FailureCategoryOK) + latest12 := now.Add(-2 * time.Minute) + createCredentialActivityAttempt(t, db, 6, 12, latest12, telemetry.FailureCategoryOK) + createCredentialActivityAttempt(t, db, 7, 13, now.Add(-time.Second), telemetry.FailureCategoryOK) + + queryCount := 0 + const callbackName = "test:credential_activity_query_count" + if err := db.Callback().Query().After("gorm:query").Register(callbackName, func(*gorm.DB) { + queryCount++ + }); err != nil { + t.Fatalf("register query counter: %v", err) + } + if err := db.Callback().Row().After("gorm:row").Register(callbackName, func(*gorm.DB) { + queryCount++ + }); err != nil { + t.Fatalf("register row query counter: %v", err) + } + t.Cleanup(func() { + _ = db.Callback().Query().Remove(callbackName) + _ = db.Callback().Row().Remove(callbackName) + }) + + result, err := service.QueryCredentialActivity(context.Background(), CredentialActivityQuery{ + CredentialIDs: []uint{11, 12}, + FromMS: from.UnixMilli(), + ToMS: now.UnixMilli(), + }) + if err != nil { + t.Fatalf("QueryCredentialActivity() error = %v", err) + } + if queryCount != 3 { + t.Fatalf("query count = %d, want one latest, one hourly, and one boundary query", queryCount) + } + if len(result) != 2 { + t.Fatalf("result = %#v, want only requested credentials", result) + } + first := result[11] + if first.CredentialID != 11 || first.SuccessCount != 6 || first.FailureCount != 6 || + !first.DataComplete || first.LastUsedAtMS == nil || *first.LastUsedAtMS != latest11.UnixMilli() { + t.Fatalf("credential 11 activity = %#v", first) + } + second := result[12] + if second.CredentialID != 12 || second.SuccessCount != 7 || second.FailureCount != 8 || + !second.DataComplete || second.LastUsedAtMS == nil || *second.LastUsedAtMS != latest12.UnixMilli() { + t.Fatalf("credential 12 activity = %#v", second) + } +} + +func TestQueryCredentialActivityReturnsZeroActivityAndIncompleteRetention(t *testing.T) { + db := openRequestLogQueryDB(t) + service := NewService(db, nil, staticRetentionPolicy{days: 0}) + now := time.Date(2026, time.August, 15, 13, 45, 0, 0, time.UTC) + service.now = func() time.Time { return now } + + result, err := service.QueryCredentialActivity(t.Context(), CredentialActivityQuery{ + CredentialIDs: []uint{21}, + FromMS: now.Add(-24 * time.Hour).UnixMilli(), + ToMS: now.UnixMilli(), + }) + if err != nil { + t.Fatalf("QueryCredentialActivity() error = %v", err) + } + activity, exists := result[21] + if !exists || activity.CredentialID != 21 || activity.SuccessCount != 0 || + activity.FailureCount != 0 || activity.LastUsedAtMS != nil || activity.DataComplete { + t.Fatalf("empty activity = %#v, exists = %v", activity, exists) + } +} + +func TestQueryCredentialActivityValidatesScope(t *testing.T) { + service := newRequestLogTestService(openRequestLogQueryDB(t)) + valid := CredentialActivityQuery{CredentialIDs: []uint{1}, FromMS: 1, ToMS: 2} + tooMany := make([]uint, 101) + for index := range tooMany { + tooMany[index] = uint(index + 1) + } + + tests := map[string]CredentialActivityQuery{ + "empty IDs": {FromMS: 1, ToMS: 2}, + "zero ID": {CredentialIDs: []uint{0}, FromMS: 1, ToMS: 2}, + "too many IDs": {CredentialIDs: tooMany, FromMS: 1, ToMS: 2}, + "negative start": {CredentialIDs: []uint{1}, FromMS: -1, ToMS: 2}, + "empty time range": {CredentialIDs: []uint{1}, FromMS: 2, ToMS: 2}, + "excessive range": {CredentialIDs: []uint{1}, FromMS: 1, ToMS: 1 + credentialActivityMaxWindowMS + 1}, + } + for name, input := range tests { + t.Run(name, func(t *testing.T) { + if _, err := service.QueryCredentialActivity(t.Context(), input); err == nil { + t.Fatalf("QueryCredentialActivity(%#v) error = nil", input) + } + }) + } + if _, err := (*Service)(nil).QueryCredentialActivity(t.Context(), valid); err == nil { + t.Fatal("nil service error = nil") + } +} + +func createCredentialAttemptStats(t *testing.T, db *gorm.DB, rows ...models.CredentialAttemptStat) { + t.Helper() + if err := db.Create(&rows).Error; err != nil { + t.Fatalf("create CredentialAttemptStats: %v", err) + } +} + +func createCredentialActivityAttempt( + t *testing.T, + db *gorm.DB, + sequence int, + credentialID uint, + completedAt time.Time, + category telemetry.FailureCategory, +) { + t.Helper() + id := fmt.Sprintf("00000000-0000-4000-8000-%012d", sequence) + row := requestLogQueryRow(id, completedAt, 1, "activity", []Attempt{{ + CredentialID: credentialID, FailureCategory: category, + }}) + row.AttemptRows[0].CompletedAtMS = completedAt.UnixMilli() + createRequestLogQueryRow(t, db, row) +} diff --git a/internal/requestlog/credential_window_usage.go b/internal/requestlog/credential_window_usage.go index 1d0e92a7..7633b9d5 100644 --- a/internal/requestlog/credential_window_usage.go +++ b/internal/requestlog/credential_window_usage.go @@ -5,7 +5,10 @@ import ( "fmt" "math" + "gorm.io/gorm" + "gpt-load/internal/platform/epochms" + "gpt-load/internal/storage/dbtx" "gpt-load/internal/storage/models" ) @@ -43,20 +46,30 @@ func (service *Service) QueryCredentialWindowUsage( if ctx == nil { ctx = context.Background() } - - var result CredentialWindowUsage - var err error switch input.Source { - case CredentialWindowUsageSourceRequestLogs: - result, err = service.queryCredentialRequestLogUsage(ctx, input) - case CredentialWindowUsageSourceHourlyStats: - result, err = service.queryCredentialHourlyUsage(ctx, input) + case CredentialWindowUsageSourceRequestLogs, CredentialWindowUsageSourceHourlyStats: default: return CredentialWindowUsage{}, fmt.Errorf( "query credential window usage: unsupported source %q", input.Source, ) } + + var result CredentialWindowUsage + err := dbtx.Run(ctx, service.db, dbtx.Options{ + Mode: dbtx.ReadSnapshot, + CleanupTimeout: usageRollbackTimeout, + Operation: "credential window usage read transaction", + }, func(connection *gorm.DB) error { + var queryErr error + switch input.Source { + case CredentialWindowUsageSourceRequestLogs: + result, queryErr = service.queryCredentialRequestLogUsage(connection, input) + case CredentialWindowUsageSourceHourlyStats: + result, queryErr = service.queryCredentialHourlyUsage(connection, input) + } + return queryErr + }) if err != nil { return CredentialWindowUsage{}, fmt.Errorf("query credential window usage: %w", err) } @@ -64,11 +77,11 @@ func (service *Service) QueryCredentialWindowUsage( } func (service *Service) queryCredentialRequestLogUsage( - ctx context.Context, + db *gorm.DB, input CredentialWindowUsageQuery, ) (CredentialWindowUsage, error) { var row credentialRequestLogUsageRow - query := service.db.WithContext(ctx).Model(&models.RequestLog{}). + query := db.Model(&models.RequestLog{}). Where("credential_id = ?", input.CredentialID). Where("completed_at_ms >= ? AND completed_at_ms < ?", input.FromMS, input.ToMS) if err := query.Select(credentialRequestLogUsageSelect).Find(&row).Error; err != nil { @@ -86,40 +99,71 @@ func (service *Service) queryCredentialRequestLogUsage( } func (service *Service) queryCredentialHourlyUsage( - ctx context.Context, + db *gorm.DB, input CredentialWindowUsageQuery, ) (CredentialWindowUsage, error) { - fromMS, err := epochms.AlignDown(input.FromMS, epochms.MillisecondsPerHour) + fullHoursFromMS, err := alignHourUp(input.FromMS) if err != nil { return CredentialWindowUsage{}, fmt.Errorf("align window start: %w", err) } - toMS, err := alignHourUp(input.ToMS) + fullHoursToMS, err := alignHourUp(input.ToMS) if err != nil { return CredentialWindowUsage{}, fmt.Errorf("align window end: %w", err) } - scope := service.db.WithContext(ctx).Model(&models.UsageStat{}). - Where("credential_id = ?", input.CredentialID). - Where("bucket_start_ms >= ? AND bucket_start_ms < ?", fromMS, toMS) - if err := validateUsageStatIntegrity(scope); err != nil { - return CredentialWindowUsage{}, err + result := CredentialWindowUsage{ + Source: CredentialWindowUsageSourceHourlyStats, + DataComplete: true, } - aggregate, err := queryUsageSummary(scope) - if err != nil { - return CredentialWindowUsage{}, err + if fullHoursFromMS < fullHoursToMS { + scope := db.Model(&models.UsageStat{}). + Where("credential_id = ?", input.CredentialID). + Where("bucket_start_ms >= ? AND bucket_start_ms < ?", fullHoursFromMS, fullHoursToMS) + if err := validateUsageStatIntegrity(scope); err != nil { + return CredentialWindowUsage{}, err + } + aggregate, err := queryUsageSummary(scope) + if err != nil { + return CredentialWindowUsage{}, err + } + result.UsageAggregate = aggregate + var latest struct { + LastUsedAtMS *int64 `gorm:"column:last_used_at_ms"` + } + if err := scope.Select("MAX(bucket_start_ms) AS last_used_at_ms").Find(&latest).Error; err != nil { + return CredentialWindowUsage{}, fmt.Errorf("query latest usage bucket: %w", err) + } + result.LastUsedAtMS = latest.LastUsedAtMS + result.DataComplete = service.hourlyWindowRetained(fullHoursFromMS) } - var latest struct { - LastUsedAtMS *int64 `gorm:"column:last_used_at_ms"` + + boundaryToMS := fullHoursFromMS + if boundaryToMS > input.ToMS { + boundaryToMS = input.ToMS } - if err := scope.Select("MAX(bucket_start_ms) AS last_used_at_ms").Find(&latest).Error; err != nil { - return CredentialWindowUsage{}, fmt.Errorf("query latest usage bucket: %w", err) + if input.FromMS < boundaryToMS { + boundary, err := service.queryCredentialRequestLogUsage(db, CredentialWindowUsageQuery{ + CredentialID: input.CredentialID, + FromMS: input.FromMS, + ToMS: boundaryToMS, + Source: CredentialWindowUsageSourceRequestLogs, + }) + if err != nil { + return CredentialWindowUsage{}, err + } + result.UsageAggregate, err = addUsageAggregates( + result.UsageAggregate, + boundary.UsageAggregate, + ) + if err != nil { + return CredentialWindowUsage{}, fmt.Errorf("merge boundary usage: %w", err) + } + result.DataComplete = result.DataComplete && boundary.DataComplete + if boundary.LastUsedAtMS != nil && + (result.LastUsedAtMS == nil || *boundary.LastUsedAtMS > *result.LastUsedAtMS) { + result.LastUsedAtMS = boundary.LastUsedAtMS + } } - return CredentialWindowUsage{ - UsageAggregate: aggregate, - Source: CredentialWindowUsageSourceHourlyStats, - DataComplete: input.FromMS == fromMS && input.ToMS == toMS && - service.hourlyWindowRetained(input.FromMS), - LastUsedAtMS: latest.LastUsedAtMS, - }, nil + return result, nil } func (service *Service) requestLogWindowRetained(fromMS int64) bool { diff --git a/internal/requestlog/credential_window_usage_test.go b/internal/requestlog/credential_window_usage_test.go index 8fc99cb5..1bc1a140 100644 --- a/internal/requestlog/credential_window_usage_test.go +++ b/internal/requestlog/credential_window_usage_test.go @@ -4,6 +4,8 @@ import ( "testing" "time" + "gorm.io/gorm" + "gpt-load/internal/pricing" "gpt-load/internal/storage/models" "gpt-load/internal/telemetry" @@ -64,6 +66,13 @@ func TestQueryCredentialWindowUsageUsesHourlyStatsForLongWindow(t *testing.T) { createUsageStats(t, db, first, last, other) from := now.Add(-7 * 24 * time.Hour).Add(10 * time.Minute) + boundary := requestLogWindowRow( + "00000000-0000-4000-8000-000000000735", + credentialID, + from.Add(5*time.Minute), + 40, + ) + createRequestLogQueryRow(t, db, boundary) result, err := service.QueryCredentialWindowUsage(t.Context(), CredentialWindowUsageQuery{ CredentialID: credentialID, FromMS: from.UnixMilli(), @@ -73,14 +82,66 @@ func TestQueryCredentialWindowUsageUsesHourlyStatsForLongWindow(t *testing.T) { if err != nil { t.Fatalf("QueryCredentialWindowUsage() error = %v", err) } - if result.Source != CredentialWindowUsageSourceHourlyStats || result.DataComplete || - result.RequestCount != 5 || result.UncachedInputTokens != 50 || - result.EstimatedCostNanoUSD != 500_000_000 || result.LastUsedAtMS == nil || + if result.Source != CredentialWindowUsageSourceHourlyStats || !result.DataComplete || + result.RequestCount != 4 || result.UncachedInputTokens != 70 || + result.EstimatedCostNanoUSD != 300_000_000 || result.LastUsedAtMS == nil || *result.LastUsedAtMS != last.BucketStartMS { t.Fatalf("hourly window usage = %#v", result) } } +func TestQueryCredentialWindowUsageUsesOneReadSnapshot(t *testing.T) { + db, dsn := openRequestLogFileDB(t) + service := newRequestLogTestService(db) + now := time.Date(2026, time.August, 14, 12, 25, 0, 0, time.UTC) + service.now = func() time.Time { return now } + credentialID := uint(42) + from := now.Add(-5 * time.Hour) + hourly := usageStat(from.Add(time.Hour).Truncate(time.Hour), 17, "codex-a", 1) + hourly.CredentialID = credentialID + createUsageStats(t, db, hourly) + + writerDB, closeWriter := openUsageQueryWriterDB(t, dsn) + defer closeWriter() + inserted := false + const callbackName = "test:credential_window_usage_snapshot_insert" + if err := db.Callback().Query().After("gorm:query").Register(callbackName, func(tx *gorm.DB) { + if inserted || tx.Statement.Table != "usage_stats" { + return + } + inserted = true + lateBoundary := requestLogWindowRow( + "00000000-0000-4000-8000-000000000736", + credentialID, + from.Add(5*time.Minute), + 40, + ) + if err := writerDB.Create(&lateBoundary).Error; err != nil { + t.Errorf("insert concurrent RequestLog: %v", err) + } + }); err != nil { + t.Fatalf("register query callback: %v", err) + } + t.Cleanup(func() { + if err := db.Callback().Query().Remove(callbackName); err != nil { + t.Errorf("remove query callback: %v", err) + } + }) + + result, err := service.QueryCredentialWindowUsage(t.Context(), CredentialWindowUsageQuery{ + CredentialID: credentialID, + FromMS: from.UnixMilli(), + ToMS: now.UnixMilli(), + Source: CredentialWindowUsageSourceHourlyStats, + }) + if err != nil { + t.Fatalf("QueryCredentialWindowUsage() error = %v", err) + } + if !inserted || result.RequestCount != 1 { + t.Fatalf("window usage did not retain one snapshot: inserted=%t result=%#v", inserted, result) + } +} + func requestLogWindowRow(id string, credentialID uint, completedAt time.Time, input int64) models.RequestLog { return models.RequestLog{ ID: id, CompletedAtMS: completedAt.UnixMilli(), AccessKeyID: 1, GroupID: 17, diff --git a/internal/requestlog/external_database_integration_test.go b/internal/requestlog/external_database_integration_test.go index 1eefdfc3..ff761bf4 100644 --- a/internal/requestlog/external_database_integration_test.go +++ b/internal/requestlog/external_database_integration_test.go @@ -14,6 +14,7 @@ import ( "gpt-load/internal/platform/redact" "gpt-load/internal/storage" "gpt-load/internal/storage/models" + "gpt-load/internal/telemetry" ) // TestExternalDatabaseRequestLogLifecycle covers the request-log write, @@ -48,6 +49,13 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) { row := aggregationRow(requestID, now.Add(-40*24*time.Hour), groupID, modelID) row.ChannelID = "openai" row.CredentialID = credentialID + row.AttemptRows = []models.RequestLogAttempt{{ + RequestID: row.ID, Sequence: 1, CompletedAtMS: row.CompletedAtMS, + GroupID: groupID, GroupName: "external-request-log", ChannelID: row.ChannelID, + CredentialID: credentialID, StatusCode: 200, + FailureCategory: string(telemetry.FailureCategoryOK), + Action: string(telemetry.ActionTerminate), + }} writer := &gormBatchWriter{db: db} if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil { t.Fatalf("first WriteBatch() error = %v", err) @@ -69,8 +77,24 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) { modelID, 1, ) + assertExternalCredentialAttemptStat(t, db, row.CompletedAtMS, credentialID, 1) service := NewService(db, redact.New(), staticRetentionPolicy{days: 1}) + service.now = func() time.Time { return now } + activity, err := service.QueryCredentialActivity(t.Context(), CredentialActivityQuery{ + CredentialIDs: []uint{credentialID}, + FromMS: row.CompletedAtMS, + ToMS: row.CompletedAtMS + int64(time.Hour/time.Millisecond), + }) + if err != nil { + t.Fatalf("QueryCredentialActivity() error = %v", err) + } + credentialActivity := activity[credentialID] + if credentialActivity.SuccessCount != 1 || credentialActivity.FailureCount != 0 || + credentialActivity.LastUsedAtMS == nil || + *credentialActivity.LastUsedAtMS != row.CompletedAtMS { + t.Fatalf("credential activity = %+v, want one exact successful attempt", credentialActivity) + } service.Sweep(context.Background(), now) assertExternalRequestLogCount(t, db, "id = ?", []any{requestID}, 0) assertExternalUsageStatCount( @@ -84,6 +108,7 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) { modelID, 0, ) + assertExternalCredentialAttemptStat(t, db, row.CompletedAtMS, credentialID, 0) var journals int64 if err := db.Model(&models.UsageAggregationJournal{}). Where("request_id = ?", requestID). @@ -95,6 +120,30 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) { } } +func assertExternalCredentialAttemptStat( + t *testing.T, + db *gorm.DB, + bucketStartMS int64, + credentialID uint, + wantRows int64, +) { + t.Helper() + var rows []models.CredentialAttemptStat + if err := db.Where( + "bucket_start_ms = ? AND credential_id = ?", + bucketStartMS, + credentialID, + ).Find(&rows).Error; err != nil { + t.Fatalf("query credential attempt stats: %v", err) + } + if int64(len(rows)) != wantRows { + t.Fatalf("credential attempt stat rows = %+v, want %d", rows, wantRows) + } + if wantRows == 1 && (rows[0].SuccessCount != 1 || rows[0].FailureCount != 0) { + t.Fatalf("credential attempt stat = %+v, want one success", rows[0]) + } +} + func assertExternalRequestLogCount(t *testing.T, db *gorm.DB, query string, args []any, want int64) { t.Helper() var count int64 diff --git a/internal/requestlog/mapper.go b/internal/requestlog/mapper.go index 803fb5db..51bae506 100644 --- a/internal/requestlog/mapper.go +++ b/internal/requestlog/mapper.go @@ -43,6 +43,17 @@ func mapEvent( } attempts := make([]models.RequestLogAttempt, 0, len(event.Attempts)) for _, attempt := range event.Attempts { + attemptCompletedAtMS := completedAtMS + if !attempt.CompletedAt.IsZero() { + attemptCompletedAtMS, err = epochms.FromTime(attempt.CompletedAt) + if err != nil { + return models.RequestLog{}, fmt.Errorf( + "map request event attempt %d completion time: %w", + attempt.Sequence, + err, + ) + } + } attemptReceipt := models.JSON(nil) if attempt.Sequence == event.Usage.AttemptSequence { attemptReceipt = receipt @@ -50,7 +61,7 @@ func mapEvent( attempts = append(attempts, models.RequestLogAttempt{ RequestID: event.RequestID, Sequence: attempt.Sequence, - CompletedAtMS: completedAtMS, + CompletedAtMS: attemptCompletedAtMS, GroupID: attempt.GroupID, GroupName: redactIdentityValue(redactor, attempt.GroupName), ChannelID: string(attempt.ChannelID), diff --git a/internal/requestlog/mapper_test.go b/internal/requestlog/mapper_test.go index 3feae2a1..e6cc2358 100644 --- a/internal/requestlog/mapper_test.go +++ b/internal/requestlog/mapper_test.go @@ -19,6 +19,7 @@ import ( func TestMapEventPersistsFrozenUsagePricingAndAttribution(t *testing.T) { event := testEvent("00000000-0000-4000-8000-000000000101") event.CompletedAt = time.Date(2026, time.July, 24, 20, 30, 0, 123, time.FixedZone("test", 8*60*60)) + event.Attempts[0].CompletedAt = event.CompletedAt.Add(-90 * time.Second) event.Protocol = protocol.OpenAICompletions event.Operation = execution.OperationChatCompletion event.ClientModel = "client-alias" @@ -83,7 +84,8 @@ func TestMapEventPersistsFrozenUsagePricingAndAttribution(t *testing.T) { } if len(row.AttemptRows) != 1 || row.AttemptRows[0].GroupID != 7 || - row.AttemptRows[0].CredentialID != 8 { + row.AttemptRows[0].CredentialID != 8 || + row.AttemptRows[0].CompletedAtMS != event.Attempts[0].CompletedAt.UnixMilli() { t.Fatalf("attempts = %+v", row.AttemptRows) } } diff --git a/internal/requestlog/retention.go b/internal/requestlog/retention.go index b35f36d1..07330748 100644 --- a/internal/requestlog/retention.go +++ b/internal/requestlog/retention.go @@ -18,7 +18,7 @@ const ( usageStatRetentionDays = 35 ) -// Sweep removes request logs, hourly usage aggregates, and aggregation journals +// Sweep removes request logs, hourly aggregates, and aggregation journals // strictly older than their respective retention boundaries. Failures are // isolated from the data plane. func (service *Service) Sweep(ctx context.Context, now time.Time) { @@ -62,9 +62,56 @@ func (service *Service) Sweep(ctx context.Context, now time.Time) { if ctx.Err() != nil { return } + service.deleteExpiredCredentialAttemptStats(ctx, usageStatCutoffMS, now) + if ctx.Err() != nil { + return + } service.deleteExpiredUsageJournals(ctx, usageStatCutoffMS, now) } +func (service *Service) deleteExpiredCredentialAttemptStats( + ctx context.Context, + cutoffMS int64, + now time.Time, +) bool { + for { + if ctx.Err() != nil { + return false + } + + var ids []uint + result := service.db.WithContext(ctx). + Model(&models.CredentialAttemptStat{}). + Where("bucket_start_ms < ?", cutoffMS). + Order("bucket_start_ms ASC"). + Order("id ASC"). + Limit(retentionBatchSize). + Pluck("id", &ids) + if result.Error != nil { + if ctx.Err() == nil { + service.recordRetentionDeleteFailure(now) + } + return false + } + if len(ids) == 0 { + return true + } + + result = service.db.WithContext(ctx). + Where("id IN ?", ids). + Delete(&models.CredentialAttemptStat{}) + if result.Error != nil { + if ctx.Err() == nil { + service.recordRetentionDeleteFailure(now) + } + return false + } + if len(ids) < retentionBatchSize { + return true + } + } +} + func retentionCutoffMS(nowMS int64, days int) (int64, error) { if days < 0 { return 0, fmt.Errorf("retention days must be non-negative") diff --git a/internal/requestlog/retention_test.go b/internal/requestlog/retention_test.go index 473dff00..bb3e6512 100644 --- a/internal/requestlog/retention_test.go +++ b/internal/requestlog/retention_test.go @@ -122,6 +122,11 @@ func TestRetentionSweepUsesFixedThirtyFiveDayIntegerUsageBoundary(t *testing.T) }).Error; err != nil { t.Fatalf("create UsageStat at %d: %v", bucketStartMS, err) } + if err := db.Create(&models.CredentialAttemptStat{ + CredentialID: uint(index + 1), BucketStartMS: bucketStartMS, SuccessCount: 1, + }).Error; err != nil { + t.Fatalf("create CredentialAttemptStat at %d: %v", bucketStartMS, err) + } } for index, item := range []struct { bucketStartMS int64 @@ -156,6 +161,18 @@ func TestRetentionSweepUsesFixedThirtyFiveDayIntegerUsageBoundary(t *testing.T) remaining[1].BucketStartMS != usageCutoffMS+3_600_000 { t.Fatalf("remaining UsageStats = %+v, want cutoff and newer integer buckets", remaining) } + var remainingAttempts []models.CredentialAttemptStat + if err := db.Order("bucket_start_ms ASC").Find(&remainingAttempts).Error; err != nil { + t.Fatalf("query remaining CredentialAttemptStats: %v", err) + } + if len(remainingAttempts) != 2 || + remainingAttempts[0].BucketStartMS != usageCutoffMS || + remainingAttempts[1].BucketStartMS != usageCutoffMS+3_600_000 { + t.Fatalf( + "remaining CredentialAttemptStats = %+v, want cutoff and newer buckets", + remainingAttempts, + ) + } var journals []models.UsageAggregationJournal if err := db.Order("request_id ASC").Find(&journals).Error; err != nil { t.Fatalf("query remaining UsageAggregationJournals: %v", err) @@ -402,6 +419,13 @@ func TestRetentionReplayBoundaryKeepsAggregationIdempotentWithoutRequestLog(t *t row := aggregationRow(aggregationRequestID(70), now.Add(-8*24*time.Hour), 15, "retention-aggregate") row.Status = string(telemetry.RequestStatusSuccess) row.UsageState = string(usage.StateComplete) + row.AttemptRows = []models.RequestLogAttempt{credentialAttemptRow( + row.ID, + 1, + now.Add(-8*24*time.Hour), + 23, + telemetry.FailureCategoryOK, + )} if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil { t.Fatalf("initial WriteBatch() error = %v", err) @@ -437,10 +461,21 @@ func assertRetentionReplayState( if err := db.Find(&stats).Error; err != nil { t.Fatalf("query UsageStats: %v", err) } + var attemptStats []models.CredentialAttemptStat + if err := db.Find(&attemptStats).Error; err != nil { + t.Fatalf("query CredentialAttemptStats: %v", err) + } if requestLogs != wantRequestLogs || len(stats) != 1 || - stats[0].RequestCount != wantAggregatedRequests { - t.Fatalf("retention replay state = logs:%d stats:%+v, want logs:%d requests:%d", - requestLogs, stats, wantRequestLogs, wantAggregatedRequests) + stats[0].RequestCount != wantAggregatedRequests || len(attemptStats) != 1 || + attemptStats[0].SuccessCount != 1 || attemptStats[0].FailureCount != 0 { + t.Fatalf( + "retention replay state = logs:%d stats:%+v attempt stats:%+v, want logs:%d requests:%d and one attempt", + requestLogs, + stats, + attemptStats, + wantRequestLogs, + wantAggregatedRequests, + ) } } diff --git a/internal/requestlog/worker.go b/internal/requestlog/worker.go index c204d018..717e4f75 100644 --- a/internal/requestlog/worker.go +++ b/internal/requestlog/worker.go @@ -168,9 +168,151 @@ func writeRequestLogBatch(tx *gorm.DB, rows []models.RequestLog) error { Order("request_id ASC").Find(&journals).Error; err != nil { return fmt.Errorf("query current usage journals: %w", err) } + if len(attemptRows) > 0 && len(journals) > 0 { + pendingRequestIDs := make(map[string]struct{}, len(journals)) + for _, journal := range journals { + pendingRequestIDs[journal.RequestID] = struct{}{} + } + pendingAttempts := make([]models.RequestLogAttempt, 0, len(attemptRows)) + for _, attempt := range attemptRows { + if _, pending := pendingRequestIDs[attempt.RequestID]; pending { + pendingAttempts = append(pendingAttempts, attempt) + } + } + if err := applyCredentialAttemptStats(tx, pendingAttempts); err != nil { + return err + } + } return applyUsageJournalBatch(tx, journals) } +type credentialAttemptStatKey struct { + CredentialID uint + BucketStartMS int64 +} + +type credentialAttemptStatDelta struct { + SuccessCount int64 + FailureCount int64 +} + +func applyCredentialAttemptStats(tx *gorm.DB, attempts []models.RequestLogAttempt) error { + deltas, err := buildCredentialAttemptStatDeltas(attempts) + if err != nil { + return err + } + if len(deltas) == 0 { + return nil + } + keys := make([]credentialAttemptStatKey, 0, len(deltas)) + credentialSet := make(map[uint]struct{}) + bucketSet := make(map[int64]struct{}) + for key := range deltas { + keys = append(keys, key) + credentialSet[key.CredentialID] = struct{}{} + bucketSet[key.BucketStartMS] = struct{}{} + } + sort.Slice(keys, func(left, right int) bool { + if keys[left].CredentialID != keys[right].CredentialID { + return keys[left].CredentialID < keys[right].CredentialID + } + return keys[left].BucketStartMS < keys[right].BucketStartMS + }) + credentialIDs := make([]uint, 0, len(credentialSet)) + for credentialID := range credentialSet { + credentialIDs = append(credentialIDs, credentialID) + } + bucketStarts := make([]int64, 0, len(bucketSet)) + for bucketStartMS := range bucketSet { + bucketStarts = append(bucketStarts, bucketStartMS) + } + + var existingRows []models.CredentialAttemptStat + if err := tx.Where("credential_id IN ? AND bucket_start_ms IN ?", credentialIDs, bucketStarts). + Find(&existingRows).Error; err != nil { + return fmt.Errorf("query credential attempt stats: %w", err) + } + existing := make(map[credentialAttemptStatKey]models.CredentialAttemptStat, len(existingRows)) + for _, row := range existingRows { + if row.SuccessCount < 0 || row.FailureCount < 0 { + return fmt.Errorf("query credential attempt stats: corrupt row") + } + existing[credentialAttemptStatKey{ + CredentialID: row.CredentialID, BucketStartMS: row.BucketStartMS, + }] = row + } + + for _, key := range keys { + row := existing[key] + row.CredentialID = key.CredentialID + row.BucketStartMS = key.BucketStartMS + delta := deltas[key] + if err := checkedInt64Add(&row.SuccessCount, delta.SuccessCount, "credential attempt success_count"); err != nil { + return err + } + if err := checkedInt64Add(&row.FailureCount, delta.FailureCount, "credential attempt failure_count"); err != nil { + return err + } + if err := tx.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "credential_id"}, {Name: "bucket_start_ms"}}, + DoUpdates: clause.AssignmentColumns([]string{ + "success_count", "failure_count", + }), + }).Create(&row).Error; err != nil { + return fmt.Errorf("upsert credential attempt stat: %w", err) + } + } + return nil +} + +func buildCredentialAttemptStatDeltas( + attempts []models.RequestLogAttempt, +) (map[credentialAttemptStatKey]credentialAttemptStatDelta, error) { + deltas := make(map[credentialAttemptStatKey]credentialAttemptStatDelta) + for _, attempt := range attempts { + if attempt.CredentialID == 0 { + return nil, fmt.Errorf("aggregate credential attempt: credential ID is zero") + } + if attempt.FailureCategory == string(telemetry.FailureCategoryDownstreamCancel) { + continue + } + bucketStartMS, err := epochms.AlignDown( + attempt.CompletedAtMS, + epochms.MillisecondsPerHour, + ) + if err != nil { + return nil, fmt.Errorf("aggregate credential attempt completion time: %w", err) + } + key := credentialAttemptStatKey{ + CredentialID: attempt.CredentialID, BucketStartMS: bucketStartMS, + } + delta := deltas[key] + switch telemetry.FailureCategory(attempt.FailureCategory) { + case telemetry.FailureCategoryOK: + if err := checkedInt64Add(&delta.SuccessCount, 1, "credential attempt success_count"); err != nil { + return nil, err + } + case telemetry.FailureCategoryRateLimited, + telemetry.FailureCategoryModelUnavailable, + telemetry.FailureCategoryInvalidKey, + telemetry.FailureCategoryUpstreamHost, + telemetry.FailureCategoryClientError, + telemetry.FailureCategoryConversionUnsupported, + telemetry.FailureCategoryAmbiguous: + if err := checkedInt64Add(&delta.FailureCount, 1, "credential attempt failure_count"); err != nil { + return nil, err + } + default: + return nil, fmt.Errorf( + "aggregate credential attempt: invalid failure category %q", + attempt.FailureCategory, + ) + } + deltas[key] = delta + } + return deltas, nil +} + func applyUsageJournalBatch( tx *gorm.DB, journals []models.UsageAggregationJournal, diff --git a/internal/requestlog/worker_test.go b/internal/requestlog/worker_test.go index 03da3ff3..7cde7cbe 100644 --- a/internal/requestlog/worker_test.go +++ b/internal/requestlog/worker_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "math" + "net/http" "reflect" "strings" "sync" @@ -542,6 +543,77 @@ func TestWriteBatchInsertsOnlyNewRequestLogsAndAggregatesUsage(t *testing.T) { } } +func TestWriteBatchAggregatesCredentialAttemptsByOwnCompletionHour(t *testing.T) { + db := openRequestLogQueryDB(t) + firstHour := time.Date(2026, time.August, 15, 14, 0, 0, 0, time.UTC) + row := aggregationRow(aggregationRequestID(90), firstHour.Add(2*time.Hour), 7, "attempt-model") + row.AttemptCount = 4 + row.AttemptRows = []models.RequestLogAttempt{ + credentialAttemptRow(row.ID, 1, firstHour.Add(time.Minute), 11, telemetry.FailureCategoryOK), + credentialAttemptRow(row.ID, 2, firstHour.Add(2*time.Minute), 11, telemetry.FailureCategoryRateLimited), + credentialAttemptRow(row.ID, 3, firstHour.Add(time.Hour+time.Minute), 22, telemetry.FailureCategoryInvalidKey), + credentialAttemptRow(row.ID, 4, firstHour.Add(time.Hour+2*time.Minute), 22, telemetry.FailureCategoryDownstreamCancel), + } + + writer := &gormBatchWriter{db: db} + if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil { + t.Fatalf("WriteBatch() error = %v", err) + } + + var stats []models.CredentialAttemptStat + if err := db.Order("credential_id ASC").Order("bucket_start_ms ASC").Find(&stats).Error; err != nil { + t.Fatalf("query CredentialAttemptStats: %v", err) + } + want := []models.CredentialAttemptStat{ + {CredentialID: 11, BucketStartMS: firstHour.UnixMilli(), SuccessCount: 1, FailureCount: 1}, + {CredentialID: 22, BucketStartMS: firstHour.Add(time.Hour).UnixMilli(), FailureCount: 1}, + } + if len(stats) != len(want) { + t.Fatalf("CredentialAttemptStat rows = %+v, want %+v", stats, want) + } + persisted := append([]models.CredentialAttemptStat(nil), stats...) + for index := range want { + stats[index].ID = 0 + if stats[index] != want[index] { + t.Fatalf("CredentialAttemptStat[%d] = %+v, want %+v", index, stats[index], want[index]) + } + } + + if err := writer.WriteBatch(context.Background(), []models.RequestLog{row}); err != nil { + t.Fatalf("replay WriteBatch() error = %v", err) + } + var replayed []models.CredentialAttemptStat + if err := db.Order("credential_id ASC").Order("bucket_start_ms ASC").Find(&replayed).Error; err != nil { + t.Fatalf("query replayed CredentialAttemptStats: %v", err) + } + if !reflect.DeepEqual(replayed, persisted) { + t.Fatalf("replay changed CredentialAttemptStats: before=%+v after=%+v", persisted, replayed) + } +} + +func credentialAttemptRow( + requestID string, + sequence int, + completedAt time.Time, + credentialID uint, + category telemetry.FailureCategory, +) models.RequestLogAttempt { + action := telemetry.ActionRetry + statusCode := http.StatusTooManyRequests + if category == telemetry.FailureCategoryOK { + action = telemetry.ActionTerminate + statusCode = http.StatusOK + } else if category == telemetry.FailureCategoryDownstreamCancel { + action = telemetry.ActionTerminate + statusCode = 499 + } + return models.RequestLogAttempt{ + RequestID: requestID, Sequence: sequence, CompletedAtMS: completedAt.UnixMilli(), + GroupID: 7, GroupName: "attempt-group", ChannelID: "openai", CredentialID: credentialID, + StatusCode: statusCode, FailureCategory: string(category), Action: string(action), + } +} + func TestWriteBatchAggregatesStatusQualityAndCompletePricedTotals(t *testing.T) { db := openRequestLogQueryDB(t) hour := time.Date(2026, time.July, 24, 13, 0, 0, 0, time.UTC) diff --git a/internal/storage/database_integration_test.go b/internal/storage/database_integration_test.go index bd560106..7a71e072 100644 --- a/internal/storage/database_integration_test.go +++ b/internal/storage/database_integration_test.go @@ -89,7 +89,7 @@ func TestExternalDatabaseLifecycle(t *testing.T) { "groups", "credentials", "access_keys", "request_logs", "request_log_attempts", "usage_aggregation_journal", "usage_stats", "model_prices", "system_settings", "jobs", "control_operations", "credential_stages", "credential_observations", - "credential_reset_operations", "schema_migrations", + "credential_reset_operations", "credential_attempt_stats", "schema_migrations", } { if !db.Migrator().HasTable(table) { t.Fatalf("table %q is missing", table) diff --git a/internal/storage/db_test.go b/internal/storage/db_test.go index f63504c1..3f0cfb98 100644 --- a/internal/storage/db_test.go +++ b/internal/storage/db_test.go @@ -602,6 +602,7 @@ func TestAutoMigrateCreatesUsageJournalAndMigrationLedger(t *testing.T) { "credential_stages", "credential_observations", "credential_reset_operations", + "credential_attempt_stats", "schema_migrations", } for _, table := range wantTables { @@ -619,6 +620,7 @@ func TestAutoMigrateCreatesUsageJournalAndMigrationLedger(t *testing.T) { "0002_subscription_connections", "0003_codex_channel", "0004_subscription_runtime", + "0005_credential_attempt_stats", } if !reflect.DeepEqual(migrationIDs, wantMigrationIDs) { t.Fatalf("schema_migrations IDs = %v, want %v", migrationIDs, wantMigrationIDs) diff --git a/internal/storage/migration.go b/internal/storage/migration.go index f3657b8d..1725a507 100644 --- a/internal/storage/migration.go +++ b/internal/storage/migration.go @@ -58,6 +58,12 @@ var migrations = []migration{ Validate: migrationfiles.Validate0004, ValidateRecoverable: migrationfiles.ValidateRecoverable0004, }, + { + ID: migrationfiles.ID0005, + Up: migrationfiles.Up0005, + Validate: migrationfiles.Validate0005, + ValidateRecoverable: migrationfiles.ValidateRecoverable0005, + }, } func applyMigrations(db *gorm.DB) error { diff --git a/internal/storage/migration_test.go b/internal/storage/migration_test.go index d74fe9f8..2de4d489 100644 --- a/internal/storage/migration_test.go +++ b/internal/storage/migration_test.go @@ -11,10 +11,16 @@ import ( ) func TestMigrationRegistryContainsOrderedMigrations(t *testing.T) { - if len(migrations) != 4 { - t.Fatalf("migration registry length = %d, want 4", len(migrations)) + if len(migrations) != 5 { + t.Fatalf("migration registry length = %d, want 5", len(migrations)) + } + wantIDs := []string{ + migrationfiles.ID0001, + migrationfiles.ID0002, + migrationfiles.ID0003, + migrationfiles.ID0004, + migrationfiles.ID0005, } - wantIDs := []string{migrationfiles.ID0001, migrationfiles.ID0002, migrationfiles.ID0003, migrationfiles.ID0004} for index, entry := range migrations { if entry.ID != wantIDs[index] || entry.Up == nil || entry.Validate == nil || entry.ValidateRecoverable == nil { diff --git a/internal/storage/migrations/0005_credential_attempt_stats.go b/internal/storage/migrations/0005_credential_attempt_stats.go new file mode 100644 index 00000000..f722be51 --- /dev/null +++ b/internal/storage/migrations/0005_credential_attempt_stats.go @@ -0,0 +1,132 @@ +package migrations + +import ( + "fmt" + "strings" + "time" + + "gorm.io/gorm" +) + +const ( + ID0005 = "0005_credential_attempt_stats" + credentialAttemptBackfillHours = 25 +) + +type credentialAttemptStat struct { + ID uint `gorm:"primaryKey;autoIncrement;index:idx_credential_attempt_stats_bucket_id,priority:2"` + CredentialID uint `gorm:"not null;check:chk_credential_attempt_stat_credential,credential_id > 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:1"` + BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_credential_attempt_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:2;index:idx_credential_attempt_stats_bucket_id,priority:1"` + SuccessCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_success_count,success_count >= 0"` + FailureCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_failure_count,failure_count >= 0"` +} + +func (credentialAttemptStat) TableName() string { return "credential_attempt_stats" } + +type credentialUsageStatIndex struct { + CredentialID uint `gorm:"column:credential_id;index:idx_usage_stats_credential_bucket,priority:1"` + BucketStartMS int64 `gorm:"column:bucket_start_ms;index:idx_usage_stats_credential_bucket,priority:2"` +} + +func (credentialUsageStatIndex) TableName() string { return "usage_stats" } + +// Up0005 creates the credential/hour attempt aggregate and seeds the window +// shown by subscription account cards from retained attempt records. +func Up0005(db *gorm.DB) error { + if err := ValidateRecoverable0005(db); err != nil { + return err + } + if err := db.AutoMigrate(&credentialAttemptStat{}); err != nil { + return fmt.Errorf("create credential attempt stats: %w", err) + } + if !db.Migrator().HasIndex("usage_stats", "idx_usage_stats_credential_bucket") { + if err := db.Migrator().CreateIndex( + &credentialUsageStatIndex{}, + "idx_usage_stats_credential_bucket", + ); err != nil { + return fmt.Errorf("create usage stat credential/time index: %w", err) + } + } + cutoffMS := time.Now().UTC().Add(-credentialAttemptBackfillHours * time.Hour).UnixMilli() + if err := backfillCredentialAttemptStats(db, cutoffMS); err != nil { + return fmt.Errorf("backfill credential attempt stats: %w", err) + } + return nil +} + +func backfillCredentialAttemptStats(db *gorm.DB, cutoffMS int64) error { + const aggregate = ` + SELECT + attempt.credential_id, + attempt.completed_at_ms - (attempt.completed_at_ms % 3600000) AS bucket_start_ms, + SUM(CASE WHEN attempt.failure_category = 'ok' THEN 1 ELSE 0 END) AS success_count, + SUM(CASE WHEN attempt.failure_category <> 'ok' THEN 1 ELSE 0 END) AS failure_count + FROM request_log_attempts AS attempt + WHERE attempt.request_id IN ( + SELECT request_log.id + FROM request_logs AS request_log + WHERE request_log.completed_at_ms >= ? + ) + AND attempt.completed_at_ms >= ? + AND attempt.failure_category <> 'downstream_cancel' + GROUP BY + attempt.credential_id, + attempt.completed_at_ms - (attempt.completed_at_ms % 3600000)` + + var statement string + switch strings.ToLower(db.Dialector.Name()) { + case "sqlite", "postgres", "postgresql": + statement = ` + INSERT INTO credential_attempt_stats ( + credential_id, bucket_start_ms, success_count, failure_count + )` + aggregate + ` + ON CONFLICT (credential_id, bucket_start_ms) DO NOTHING` + case "mysql": + statement = ` + INSERT INTO credential_attempt_stats ( + credential_id, bucket_start_ms, success_count, failure_count + )` + aggregate + ` + ON DUPLICATE KEY UPDATE + credential_id = credential_attempt_stats.credential_id` + default: + return fmt.Errorf("unsupported database driver %q", db.Dialector.Name()) + } + return db.Exec(statement, cutoffMS, cutoffMS).Error +} + +func ValidateRecoverable0005(db *gorm.DB) error { + for _, table := range []string{"request_logs", "request_log_attempts", "usage_stats"} { + if !db.Migrator().HasTable(table) { + return fmt.Errorf("required credential activity source table %q is missing", table) + } + } + return nil +} + +func Validate0005(db *gorm.DB) error { + if err := ValidateRecoverable0005(db); err != nil { + return err + } + if !db.Migrator().HasTable(&credentialAttemptStat{}) { + return fmt.Errorf("validate credential attempt stats schema: table is missing") + } + for _, column := range []string{ + "id", "credential_id", "bucket_start_ms", "success_count", "failure_count", + } { + if !db.Migrator().HasColumn(&credentialAttemptStat{}, column) { + return fmt.Errorf("validate credential attempt stats schema: column %q is missing", column) + } + } + for _, index := range []string{ + "idx_credential_attempt_stats_identity", + "idx_credential_attempt_stats_bucket_id", + } { + if !db.Migrator().HasIndex(&credentialAttemptStat{}, index) { + return fmt.Errorf("validate credential attempt stats schema: index %q is missing", index) + } + } + if !db.Migrator().HasIndex("usage_stats", "idx_usage_stats_credential_bucket") { + return fmt.Errorf("validate credential attempt stats schema: usage stat credential/time index is missing") + } + return nil +} diff --git a/internal/storage/migrations/0005_credential_attempt_stats_test.go b/internal/storage/migrations/0005_credential_attempt_stats_test.go new file mode 100644 index 00000000..0bede06f --- /dev/null +++ b/internal/storage/migrations/0005_credential_attempt_stats_test.go @@ -0,0 +1,191 @@ +package migrations_test + +import ( + "reflect" + "testing" + "time" + + "gorm.io/gorm" + + "gpt-load/internal/pricing" + "gpt-load/internal/storage/migrations" + "gpt-load/internal/storage/models" + "gpt-load/internal/telemetry" + "gpt-load/internal/usage" +) + +func TestCredentialAttemptStatsMigrationCreatesSchemaAndBackfillsRecentAttempts(t *testing.T) { + db := openInitialTestDatabase(t) + applyCredentialAttemptStatsPrerequisites(t, db) + + now := time.Now().UTC().Truncate(time.Hour) + firstHour := now.Add(-2 * time.Hour) + secondHour := now.Add(-time.Hour) + createAttemptMigrationRequest(t, db, "00000000-0000-4000-8000-000000000501", firstHour, []attemptMigrationInput{ + {credentialID: 11, failureCategory: telemetry.FailureCategoryOK}, + {credentialID: 11, failureCategory: telemetry.FailureCategoryRateLimited}, + {credentialID: 12, failureCategory: telemetry.FailureCategoryInvalidKey}, + {credentialID: 13, failureCategory: telemetry.FailureCategoryDownstreamCancel}, + }) + createAttemptMigrationRequest(t, db, "00000000-0000-4000-8000-000000000502", secondHour, []attemptMigrationInput{ + {credentialID: 11, failureCategory: telemetry.FailureCategoryOK}, + }) + createAttemptMigrationRequest(t, db, "00000000-0000-4000-8000-000000000503", now.Add(-26*time.Hour), []attemptMigrationInput{ + {credentialID: 14, failureCategory: telemetry.FailureCategoryOK}, + }) + + if err := migrations.Up0005(db); err != nil { + t.Fatalf("Up0005() error = %v", err) + } + if err := migrations.Validate0005(db); err != nil { + t.Fatalf("Validate0005() error = %v", err) + } + + assertColumns(t, db, "credential_attempt_stats", []string{ + "id", "credential_id", "bucket_start_ms", "success_count", "failure_count", + }) + assertSQLiteIndex(t, db, "credential_attempt_stats", "idx_credential_attempt_stats_identity", true, []string{ + "credential_id", "bucket_start_ms", + }) + assertSQLiteIndex(t, db, "credential_attempt_stats", "idx_credential_attempt_stats_bucket_id", false, []string{ + "bucket_start_ms", "id", + }) + assertSQLiteIndex(t, db, "usage_stats", "idx_usage_stats_credential_bucket", false, []string{ + "credential_id", "bucket_start_ms", + }) + + type aggregate struct { + CredentialID uint + BucketStartMS int64 + SuccessCount int64 + FailureCount int64 + } + var got []aggregate + if err := db.Table("credential_attempt_stats"). + Order("credential_id ASC"). + Order("bucket_start_ms ASC"). + Find(&got).Error; err != nil { + t.Fatalf("query credential attempt stats: %v", err) + } + want := []aggregate{ + {CredentialID: 11, BucketStartMS: firstHour.UnixMilli(), SuccessCount: 1, FailureCount: 1}, + {CredentialID: 11, BucketStartMS: secondHour.UnixMilli(), SuccessCount: 1, FailureCount: 0}, + {CredentialID: 12, BucketStartMS: firstHour.UnixMilli(), SuccessCount: 0, FailureCount: 1}, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("credential attempt stats = %#v, want %#v", got, want) + } + + if err := migrations.Up0005(db); err != nil { + t.Fatalf("repeat Up0005() error = %v", err) + } + var repeated []aggregate + if err := db.Table("credential_attempt_stats"). + Order("credential_id ASC"). + Order("bucket_start_ms ASC"). + Find(&repeated).Error; err != nil { + t.Fatalf("query repeated credential attempt stats: %v", err) + } + if !reflect.DeepEqual(repeated, want) { + t.Fatalf("repeated credential attempt stats = %#v, want %#v", repeated, want) + } +} + +type attemptMigrationInput struct { + credentialID uint + failureCategory telemetry.FailureCategory +} + +func applyCredentialAttemptStatsPrerequisites(t *testing.T, db *gorm.DB) { + t.Helper() + for id, up := range []struct { + id string + up func(*gorm.DB) error + }{ + {id: migrations.ID0001, up: migrations.Up0001}, + {id: migrations.ID0002, up: migrations.Up0002}, + {id: migrations.ID0003, up: migrations.Up0003}, + {id: migrations.ID0004, up: migrations.Up0004}, + } { + if err := up.up(db); err != nil { + t.Fatalf("apply prerequisite %d (%s): %v", id, up.id, err) + } + } +} + +func createAttemptMigrationRequest( + t *testing.T, + db *gorm.DB, + requestID string, + completedAt time.Time, + inputs []attemptMigrationInput, +) { + t.Helper() + request := models.RequestLog{ + ID: requestID, CompletedAtMS: completedAt.UnixMilli(), AccessKeyID: 1, + GroupID: 1, ChannelID: "openai", CredentialID: inputs[len(inputs)-1].credentialID, + Protocol: "openai-completions", ClientModel: "migration-model", UpstreamModel: "migration-model", + ModelConsistency: string(telemetry.ModelConsistencyMatch), Status: string(telemetry.RequestStatusSuccess), + StatusCode: 200, DurationMs: 1, AttemptCount: len(inputs), + UsageState: string(usage.StateNotApplicable), CostState: string(pricing.CostStateNotApplicable), + PricingCompleteness: string(pricing.CompletenessNotApplicable), + } + if err := db.Create(&request).Error; err != nil { + t.Fatalf("create request log %s: %v", requestID, err) + } + for index, input := range inputs { + attempt := models.RequestLogAttempt{ + RequestID: requestID, Sequence: index + 1, CompletedAtMS: completedAt.UnixMilli(), + GroupID: 1, GroupName: "migration-group", ChannelID: "openai", CredentialID: input.credentialID, + StatusCode: 200, DurationMs: 1, FailureCategory: string(input.failureCategory), + Action: string(telemetry.ActionTerminate), ErrorSummary: "", + } + if err := db.Create(&attempt).Error; err != nil { + t.Fatalf("create request attempt %s/%d: %v", requestID, index+1, err) + } + } +} + +func assertSQLiteIndex( + t *testing.T, + db *gorm.DB, + table string, + indexName string, + wantUnique bool, + wantColumns []string, +) { + t.Helper() + var indexes []struct { + Name string + Unique int + } + if err := db.Raw("PRAGMA index_list('" + table + "')").Scan(&indexes).Error; err != nil { + t.Fatalf("inspect %s indexes: %v", table, err) + } + for _, index := range indexes { + if index.Name != indexName { + continue + } + if (index.Unique == 1) != wantUnique { + t.Fatalf("%s unique = %d, want %t", indexName, index.Unique, wantUnique) + } + var columns []struct { + Name string + Key int + } + if err := db.Raw("PRAGMA index_xinfo('" + indexName + "')").Scan(&columns).Error; err != nil { + t.Fatalf("inspect %s columns: %v", indexName, err) + } + gotColumns := make([]string, 0, len(wantColumns)) + for _, column := range columns { + if column.Key == 1 { + gotColumns = append(gotColumns, column.Name) + } + } + if !reflect.DeepEqual(gotColumns, wantColumns) { + t.Fatalf("%s columns = %v, want %v", indexName, gotColumns, wantColumns) + } + return + } + t.Fatalf("%s is missing", indexName) +} diff --git a/internal/storage/models/request.go b/internal/storage/models/request.go index f1544c9f..f73e6b92 100644 --- a/internal/storage/models/request.go +++ b/internal/storage/models/request.go @@ -72,6 +72,17 @@ type RequestLogAttempt struct { RequestLog *RequestLog `gorm:"foreignKey:RequestID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:CASCADE"` } +// CredentialAttemptStat is the hourly account-level outcome aggregate used by +// subscription account cards. Downstream cancellations are intentionally not +// included in either outcome count. +type CredentialAttemptStat struct { + ID uint `gorm:"primaryKey;autoIncrement;index:idx_credential_attempt_stats_bucket_id,priority:2"` + CredentialID uint `gorm:"not null;check:chk_credential_attempt_stat_credential,credential_id > 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:1"` + BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_credential_attempt_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_credential_attempt_stats_identity,priority:2;index:idx_credential_attempt_stats_bucket_id,priority:1"` + SuccessCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_success_count,success_count >= 0"` + FailureCount int64 `gorm:"not null;default:0;check:chk_credential_attempt_stat_failure_count,failure_count >= 0"` +} + // UsageAggregationJournal is the request-idempotent input for hourly usage // aggregation. It is staged, applied, and committed in the same transaction as // its RequestLog, and intentionally excludes request and error payloads. @@ -108,11 +119,11 @@ func (UsageAggregationJournal) TableName() string { // credential, and upstream model. type UsageStat struct { ID uint `gorm:"primaryKey;autoIncrement"` - BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_usage_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_usage_stats_identity,priority:1"` + BucketStartMS int64 `gorm:"column:bucket_start_ms;not null;check:chk_usage_stat_bucket,bucket_start_ms >= 0;uniqueIndex:idx_usage_stats_identity,priority:1;index:idx_usage_stats_credential_bucket,priority:2"` AccessKeyID uint `gorm:"not null;uniqueIndex:idx_usage_stats_identity,priority:2"` ChannelID string `gorm:"type:varchar(64);not null;default:'';uniqueIndex:idx_usage_stats_identity,priority:3"` GroupID uint `gorm:"not null;uniqueIndex:idx_usage_stats_identity,priority:4"` - CredentialID uint `gorm:"not null;default:0;uniqueIndex:idx_usage_stats_identity,priority:5"` + CredentialID uint `gorm:"not null;default:0;uniqueIndex:idx_usage_stats_identity,priority:5;index:idx_usage_stats_credential_bucket,priority:1"` Model string `gorm:"type:varchar(255);not null;uniqueIndex:idx_usage_stats_identity,priority:6"` RequestCount int64 `gorm:"not null;default:0;check:chk_usage_stat_request_count,request_count >= 0;check:chk_usage_stat_request_outcome,request_count = success_count + failure_count"` SuccessCount int64 `gorm:"not null;default:0;check:chk_usage_stat_success_count,success_count >= 0"` diff --git a/internal/telemetry/requestlog.go b/internal/telemetry/requestlog.go index 509116fe..8bb69612 100644 --- a/internal/telemetry/requestlog.go +++ b/internal/telemetry/requestlog.go @@ -54,6 +54,7 @@ const ( type Attempt struct { Sequence int + CompletedAt time.Time GroupID uint GroupName string ChannelID channel.ID diff --git a/internal/telemetry/requestlog_test.go b/internal/telemetry/requestlog_test.go index 3093554e..48f363f6 100644 --- a/internal/telemetry/requestlog_test.go +++ b/internal/telemetry/requestlog_test.go @@ -40,6 +40,7 @@ func TestRequestTelemetryContractUsesExactFieldAllowlist(t *testing.T) { }, reflect.TypeOf(Attempt{}): { "Sequence", + "CompletedAt", "GroupID", "GroupName", "ChannelID", diff --git a/web/src/api/control/types.ts b/web/src/api/control/types.ts index b8366008..96161eff 100644 --- a/web/src/api/control/types.ts +++ b/web/src/api/control/types.ts @@ -219,19 +219,25 @@ export interface CredentialItemDto { last_failure_category: FailureCategory last_status_code: number | null cooldown_until_ms: number | null + last_used_at_ms?: number daily_usage?: CredentialDailyUsageDto recovery: CredentialRecoveryDto } -/** 固定 24 小时窗口的请求结果分布，来源是请求日志而非 health 的 5 分钟内存窗口。 */ +/** 固定 24 小时窗口的上游尝试结果分布，来源是小时聚合而非 health 的 5 分钟内存窗口。 */ export interface CredentialDailyUsageDto { window_seconds: number success_count: number failure_count: number - /** false 表示请求日志留存期短于该窗口，计数偏低。 */ + /** false 表示统计数据未覆盖完整窗口，计数可能偏低。 */ data_complete: boolean } +export interface CredentialDetailDto { + credential: CredentialItemDto + observation: CredentialObservationDto +} + export interface CredentialRevealDto { credential_id: number credential: Record diff --git a/web/src/app/resources/credentials.ts b/web/src/app/resources/credentials.ts index 1810fb1c..4359bcdb 100644 --- a/web/src/app/resources/credentials.ts +++ b/web/src/app/resources/credentials.ts @@ -13,6 +13,7 @@ import type { CredentialCollectionFilters, CredentialConfiguredStatus, CredentialDailyUsageDto, + CredentialDetailDto, CredentialItemDto, CredentialObservationDto, CredentialObservationSnapshotDto, @@ -47,6 +48,7 @@ export type { CredentialCollectionFilters, CredentialConfiguredStatus, CredentialDailyUsageDto, + CredentialDetailDto, CredentialItemDto, CredentialRecoveryDto, CredentialRevealDto, @@ -98,9 +100,11 @@ const credentialItemFields = [ 'last_failure_category', 'last_status_code', 'cooldown_until_ms', + 'last_used_at_ms', 'daily_usage', 'recovery', ] as const +const credentialDetailFields = ['credential', 'observation'] as const const credentialDailyUsageFields = [ 'window_seconds', 'success_count', @@ -450,6 +454,9 @@ export function projectCredentialItem(value: unknown): CredentialItemDto { ? null : projectSafeInteger(record.last_status_code, { minimum: 100, maximum: 999 }), cooldown_until_ms: cooldownUntil, + ...(record.last_used_at_ms === undefined + ? {} + : { last_used_at_ms: projectEpochMilliseconds(record.last_used_at_ms) }), ...(record.daily_usage === undefined ? {} : { daily_usage: projectDailyUsage(record.daily_usage) }), @@ -468,6 +475,15 @@ function projectDailyUsage(value: unknown): CredentialDailyUsageDto { } } +export function projectCredentialDetail(value: unknown): CredentialDetailDto { + const record = projectRecord(value) + assertNoSecretLikeFields(record, credentialDetailFields) + return { + credential: projectCredentialItem(record.credential), + observation: projectObservation(record.observation), + } +} + function projectPagination(value: unknown): CredentialCollectionDto['pagination'] { const record = projectRecord(value) assertNoSecretLikeFields(record, credentialPaginationFields) @@ -570,6 +586,22 @@ export async function getCredentialCollection( return result } +export async function getCredentialDetail( + client: ApiClient, + groupId: number, + credentialId: number, + signal?: AbortSignal, +): Promise { + const result = projectCredentialDetail( + await client.request(`/api/groups/${groupId}/credentials/${credentialId}`, { + method: 'GET', + signal, + }), + ) + if (result.credential.credential_id !== credentialId) invalidResponse() + return result +} + const manualGroupQueryOptions = { refetchOnWindowFocus: false, refetchOnReconnect: false, diff --git a/web/src/features/groups/credentials/GroupCredentialsTab.vue b/web/src/features/groups/credentials/GroupCredentialsTab.vue index e8003dc0..2a677df6 100644 --- a/web/src/features/groups/credentials/GroupCredentialsTab.vue +++ b/web/src/features/groups/credentials/GroupCredentialsTab.vue @@ -22,6 +22,7 @@ import { cacheCredentialItem, consumeCredentialResetCredit, credentialCollectionQueryOptions, + getCredentialDetail, revealCredential, restoreCredential, refreshCredentialObservation, @@ -347,6 +348,52 @@ function cachedCurrentSummary(): CredentialSummaryDto | undefined { )?.summary } +function cachedCurrentCredential(id: number): CredentialItemDto | undefined { + return queryClient + .getQueryData( + controlQueryKeys.groups.credentials(props.groupId, filters.value), + ) + ?.items.find(({ credential_id }) => credential_id === id) +} + +function withPreservedObservedUsage( + item: CredentialItemDto, + sourceObservation: CredentialItemDto['observation'], +): CredentialItemDto { + const targetObservation = item.observation + if ( + !targetObservation?.snapshot || + !sourceObservation?.snapshot || + targetObservation.observation_version !== sourceObservation.observation_version || + targetObservation.observed_at_ms !== sourceObservation.observed_at_ms + ) { + return item + } + + const observedUsageByWindow = new Map( + sourceObservation.snapshot.quota_windows.flatMap((window) => + window.observed_usage === undefined ? [] : [[window.id, window.observed_usage] as const], + ), + ) + if (observedUsageByWindow.size === 0) return item + + return { + ...item, + observation: { + ...targetObservation, + snapshot: { + ...targetObservation.snapshot, + quota_windows: targetObservation.snapshot.quota_windows.map((window) => { + const observedUsage = observedUsageByWindow.get(window.id) + return window.observed_usage !== undefined || observedUsage === undefined + ? window + : { ...window, observed_usage: observedUsage } + }), + }, + }, + } +} + function synchronizeGroupSummary(summary: CredentialSummaryDto | undefined): void { const queryKey = controlQueryKeys.groups.summary(props.groupId) if (summary === undefined) { @@ -400,8 +447,21 @@ async function refetchActiveCredentialPage(): Promise { async function reconcileItem(result: CredentialItemDto, refetchActive: boolean): Promise { try { - await cacheCredentialItem(queryClient, props.groupId, result) - if (refetchActive) await refetchActiveCredentialPage() + const current = cachedCurrentCredential(result.credential_id) + const reconciled = + current === undefined ? result : withPreservedObservedUsage(result, current.observation) + await cacheCredentialItem(queryClient, props.groupId, reconciled) + if (refetchActive) { + await refetchActiveCredentialPage() + const refreshed = cachedCurrentCredential(result.credential_id) + if (refreshed !== undefined) { + await cacheCredentialItem( + queryClient, + props.groupId, + withPreservedObservedUsage(refreshed, reconciled.observation), + ) + } + } const summary = cachedCurrentSummary() if (summary === undefined) { await invalidateReconciliationQueries() @@ -424,8 +484,7 @@ async function refreshObservation(item: CredentialItemDto): Promise { props.groupId, item.credential_id, ) - const next = { ...item, observation } - await reconcileItem(next, false) + await reconcileItem({ ...item, observation }, true) } catch (cause) { feedback.value = t( presentSubscriptionErrorKey(cause, 'group.credentials.subscription.syncFailed'), @@ -435,6 +494,19 @@ async function refreshObservation(item: CredentialItemDto): Promise { } } +async function loadCredentialUsage(item: CredentialItemDto): Promise { + if (pending(item.credential_id)) return + setPending(item.credential_id, 'usage', true) + try { + const detail = await getCredentialDetail(client, props.groupId, item.credential_id) + await cacheCredentialItem(queryClient, props.groupId, detail.credential) + } catch { + feedback.value = t('group.credentials.loadFailed') + } finally { + setPending(item.credential_id, 'usage', false) + } +} + function openResetCreditDialog(item: CredentialItemDto): void { if (pending(item.credential_id)) return const idempotencyKey = resetOperationKeys.get(item.credential_id) ?? crypto.randomUUID() @@ -944,6 +1016,7 @@ async function runBatch( @toggle="mutateItem($event, 'toggle')" @restore="mutateItem($event, 'restore')" @refresh="refreshObservation" + @load-usage="loadCredentialUsage" @reset="openResetCreditDialog" @reauthorize="openConnectionWorkspace" @remove="deleteTarget = { ids: [$event.credential_id], mask: $event.mask }" diff --git a/web/src/features/groups/credentials/SubscriptionAccountCard.vue b/web/src/features/groups/credentials/SubscriptionAccountCard.vue index 8c908b6c..9bfc237c 100644 --- a/web/src/features/groups/credentials/SubscriptionAccountCard.vue +++ b/web/src/features/groups/credentials/SubscriptionAccountCard.vue @@ -22,6 +22,7 @@ const emit = defineEmits<{ toggle: [item: CredentialItemDto] restore: [item: CredentialItemDto] refresh: [item: CredentialItemDto] + 'load-usage': [item: CredentialItemDto] reset: [item: CredentialItemDto] reauthorize: [item: CredentialItemDto] remove: [item: CredentialItemDto] @@ -54,12 +55,7 @@ const quotaWindows = computed(() => const accountQuotaWindows = computed(() => quotaWindows.value.filter((window) => window.scope === 'account'), ) -const lastUsedAtMS = computed(() => { - const values = accountQuotaWindows.value - .map((window) => window.observed_usage?.last_used_at_ms) - .filter((value): value is number => value !== undefined) - return values.length ? Math.max(...values) : undefined -}) +const lastUsedAtMS = computed(() => props.item.last_used_at_ms) const constrainedModels = computed(() => Array.from(new Set(quotaWindows.value.flatMap((window) => window.model_ids ?? []))), ) @@ -152,11 +148,11 @@ const authIssue = computed(() => { const key = props.item.auth_error_code ? authErrorKeys[props.item.auth_error_code] : undefined return key ? t(key) : t(`group.credentials.subscription.auth.${props.item.auth_state}`) }) -// 卡片给的是请求日志里的 24 小时结果分布，不是 health 那个 5 分钟内存窗口 +// 卡片给的是按账号上游尝试聚合的 24 小时结果分布，不是 health 那个 5 分钟内存窗口 //（后者重启即清零，是给调度判定用的）。始终并排给出成功与失败：失败 0 本身是 // 信息，也让这一行宽度稳定。 const dailyUsage = computed(() => props.item.daily_usage) -// 请求日志留存短于 24 小时时计数会偏低，只用 title 提示，不占版面。 +// 小时统计未覆盖完整 24 小时时计数会偏低，只用 title 提示，不占版面。 const dailyIncompleteHint = computed(() => dailyUsage.value && !dailyUsage.value.data_complete ? t('group.credentials.subscription.dailyIncomplete') @@ -179,6 +175,65 @@ function observationErrorLabel(code: string | undefined): string { return key ? t(key) : code } +function windowUsageCanLoad(window: CredentialQuotaWindowDto): boolean { + if (!observationIsCurrent.value) return false + const resetAtMS = window.reset_at_ms + const windowSeconds = window.window_seconds + if ( + resetAtMS === undefined || + windowSeconds === undefined || + windowSeconds <= 0 || + resetAtMS <= nowMs.value + ) { + return false + } + const windowMS = windowSeconds * 1_000 + return ( + Number.isSafeInteger(windowMS) && windowMS <= resetAtMS && resetAtMS - windowMS < nowMs.value + ) +} + +function loadUsageOnOpen(event: Event): void { + const details = event.currentTarget as HTMLDetailsElement + if ( + details.open && + accountQuotaWindows.value.some( + (window) => windowUsageCanLoad(window) && window.observed_usage === undefined, + ) + ) { + emit('load-usage', props.item) + } +} + +function estimateTitles(...keys: string[]): string | undefined { + const titles = keys.filter(Boolean).map((key) => t(key)) + return titles.length === 0 ? undefined : titles.join(' · ') +} + +function requestCountTitle(window: CredentialQuotaWindowDto): string | undefined { + return window.observed_usage?.data_complete === false + ? t('group.credentials.subscription.estimate.dataIncomplete') + : undefined +} + +function tokenCountTitle(window: CredentialQuotaWindowDto): string | undefined { + const observed = window.observed_usage + if (!observed) return undefined + return estimateTitles( + observed.data_complete ? '' : 'group.credentials.subscription.estimate.dataIncomplete', + observed.usage_complete ? '' : 'group.credentials.subscription.estimate.usageIncomplete', + ) +} + +function referenceCostTitle(window: CredentialQuotaWindowDto): string | undefined { + const observed = window.observed_usage + if (!observed) return undefined + return estimateTitles( + observed.data_complete ? '' : 'group.credentials.subscription.estimate.dataIncomplete', + observed.pricing_complete ? '' : 'group.credentials.subscription.estimate.pricingIncomplete', + ) +} + function remainingPercent(window: CredentialQuotaWindowDto): number | undefined { if (!quotaWindowIsCurrent(window)) return undefined if (window.utilization !== undefined) return Math.round((1 - window.utilization) * 100) @@ -390,10 +445,10 @@ function runMenuAction(action: 'reauthorize' | 'toggle' | 'restore' | 'remove'): - -