Merge pull request #4850 from alfadb/feat/ollama-usage-request-debounce

feat(ollama): 按模型请求刷新云端用量
This commit is contained in:
Wesley Liddick
2026-07-25 18:09:21 +08:00
committed by GitHub
19 changed files with 849 additions and 130 deletions
@@ -61,7 +61,7 @@ func (r *ollamaCloudUsageHandlerTestRepo) UpdateOllamaCloudUsageSnapshot(context
func (r *ollamaCloudUsageHandlerTestRepo) DisableOllamaCloudUsageAutoRefresh(context.Context, *service.Account) error {
return nil
}
func (r *ollamaCloudUsageHandlerTestRepo) ListDueOllamaCloudUsageAccounts(context.Context, time.Time, int) ([]service.Account, error) {
func (r *ollamaCloudUsageHandlerTestRepo) ListDueOllamaCloudUsageAccounts(context.Context, time.Time, time.Duration, time.Duration, int) ([]service.Account, error) {
return nil, nil
}
@@ -273,4 +273,5 @@ func TestGetOllamaCloudUsageSettingsHandlerSuccess(t *testing.T) {
require.Equal(t, http.StatusOK, recorder.Code)
require.Contains(t, recorder.Body.String(), `"enabled":false`)
require.Contains(t, recorder.Body.String(), `"interval_minutes":60`)
require.Contains(t, recorder.Body.String(), `"debounce_minutes":1`)
}
@@ -354,74 +354,160 @@ func canonicalJSON(raw string) string {
return string(encoded)
}
// ListDueOllamaCloudUsageAccounts returns at most one due representative per
// exact API key before hydration, preventing one shared group from consuming a
// whole runner cycle.
func (r *accountRepository) ListDueOllamaCloudUsageAccounts(ctx context.Context, now time.Time, limit int) ([]service.Account, error) {
// ollamaCloudUsageParseRFC3339SQL reuses the verified RFC3339(/Nano) parse path
// for a snapshot timestamp expression. Invalid or missing values fail open to NULL.
func ollamaCloudUsageParseRFC3339SQL(expression string) string {
return `CASE
WHEN ` + expression + ` IS NULL THEN NULL
WHEN ` + expression + ` ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:[0-9]{2})$'
THEN jsonb_path_query_first_tz(
to_jsonb(regexp_replace(
` + expression + `,
'(\.[0-9]{6})[0-9]+(Z|[+-][0-9]{2}:[0-9]{2})$',
'\1\2'
)),
'$.datetime()', '{}'::jsonb, true
) #>> '{}'
ELSE NULL
END`
}
// ListDueOllamaCloudUsageAccounts returns at most one truly-due activity-driven
// candidate per exact API key. Due timing (debounce, max-wait, failure backoff)
// is evaluated in SQL before LIMIT so non-due active groups cannot starve due ones.
// Account.LastUsedAt is stamped with the group MAX(last_used_at) for a service
// pure-function recheck against races between list and refresh.
//
// Rules mirror service.ollamaCloudUsageAutoRefreshDueAt:
// - missing/invalid snapshot or times → fail-open first due
// - success: activity after fetched_at; due_at = LEAST(last_used+debounce, fetched+maxWait)
// - failed/unauthorized: activity after last_attempt; activity_due = LEAST(...);
// final due_at is not earlier than a valid next_refresh_at (invalid/missing fail-open)
func (r *accountRepository) ListDueOllamaCloudUsageAccounts(
ctx context.Context,
now time.Time,
debounce, maxWait time.Duration,
limit int,
) ([]service.Account, error) {
if limit <= 0 {
return []service.Account{}, nil
}
if r == nil || r.sql == nil {
return nil, errors.New("account repository SQL executor not configured")
}
if debounce <= 0 {
debounce = time.Minute
}
if maxWait <= 0 {
maxWait = time.Hour
}
debounceSeconds := debounce.Seconds()
maxWaitSeconds := maxWait.Seconds()
rows, err := r.sql.QueryContext(ctx, `
WITH candidates AS (
SELECT id, credentials ->> 'api_key' AS api_key,
extra #>> '{ollama_cloud_usage_snapshot,next_refresh_at}' AS next_refresh_at
WITH eligible AS (
SELECT id,
credentials ->> 'api_key' AS api_key,
last_used_at,
extra -> 'ollama_cloud_usage_snapshot' AS snapshot
FROM accounts
WHERE deleted_at IS NULL
AND status = 'active'
AND `+ollamaCloudUsageEligibleSQL+`
AND jsonb_typeof(extra -> 'ollama_cloud_usage_session') = 'string'
AND extra @> '{"ollama_cloud_usage_auto_refresh": true}'::jsonb
), group_activity AS (
SELECT credentials ->> 'api_key' AS api_key,
MAX(last_used_at) AS group_last_used_at
FROM accounts
WHERE deleted_at IS NULL
AND `+ollamaCloudUsageEligibleSQL+`
AND jsonb_typeof(credentials -> 'api_key') = 'string'
GROUP BY credentials ->> 'api_key'
), joined AS (
SELECT e.id, e.api_key, e.snapshot, g.group_last_used_at,
e.snapshot #>> '{status}' AS status,
e.snapshot #>> '{fetched_at}' AS fetched_at,
e.snapshot #>> '{last_attempt_at}' AS last_attempt_at,
e.snapshot #>> '{next_refresh_at}' AS next_refresh_at
FROM eligible e
JOIN group_activity g ON g.api_key = e.api_key
), parsed AS MATERIALIZED (
SELECT id, api_key, next_refresh_at,
next_refresh_at ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]+)?(Z|[+-][0-9]{2}:[0-9]{2})$' AS rfc3339_shape,
jsonb_path_query_first_tz(
to_jsonb(regexp_replace(
next_refresh_at,
'(\.[0-9]{6})[0-9]+(Z|[+-][0-9]{2}:[0-9]{2})$',
'\1\2'
)),
'$.datetime()', '{}'::jsonb, true
) #>> '{}' AS parsed_next_refresh_at
FROM candidates
), due AS (
SELECT id, api_key, snapshot, group_last_used_at, status,
`+ollamaCloudUsageParseRFC3339SQL("fetched_at")+` AS parsed_fetched_at,
`+ollamaCloudUsageParseRFC3339SQL("last_attempt_at")+` AS parsed_last_attempt_at,
`+ollamaCloudUsageParseRFC3339SQL("next_refresh_at")+` AS parsed_next_refresh_at
FROM joined
), timed AS (
SELECT *,
CASE WHEN next_refresh_at IS NULL OR NOT rfc3339_shape OR parsed_next_refresh_at IS NULL THEN 0 ELSE 1 END AS due_class
CASE
WHEN status = 'ok'
AND parsed_fetched_at IS NOT NULL
AND group_last_used_at IS NOT NULL
AND group_last_used_at > parsed_fetched_at::timestamptz
THEN LEAST(
group_last_used_at + make_interval(secs => $2::double precision),
parsed_fetched_at::timestamptz + make_interval(secs => $3::double precision)
)
WHEN status IN ('failed', 'unauthorized')
AND parsed_last_attempt_at IS NOT NULL
AND group_last_used_at IS NOT NULL
AND group_last_used_at > parsed_last_attempt_at::timestamptz
THEN GREATEST(
LEAST(
group_last_used_at + make_interval(secs => $2::double precision),
parsed_last_attempt_at::timestamptz + make_interval(secs => $3::double precision)
),
COALESCE(parsed_next_refresh_at::timestamptz, '-infinity'::timestamptz)
)
ELSE NULL
END AS activity_due_at
FROM parsed
WHERE next_refresh_at IS NULL
OR NOT rfc3339_shape
OR parsed_next_refresh_at IS NULL
OR parsed_next_refresh_at::timestamptz <= $1
), candidates AS (
SELECT *,
CASE
WHEN snapshot IS NULL OR snapshot = 'null'::jsonb OR status IS NULL
OR status NOT IN ('ok', 'failed', 'unauthorized') THEN 0
WHEN status = 'ok' AND parsed_fetched_at IS NULL THEN 0
WHEN status IN ('failed', 'unauthorized') AND parsed_last_attempt_at IS NULL THEN 0
WHEN activity_due_at IS NOT NULL AND $1 >= activity_due_at THEN 1
ELSE NULL
END AS due_class,
activity_due_at AS due_at
FROM timed
), ranked AS (
SELECT *, row_number() OVER (
PARTITION BY api_key
ORDER BY due_class,
CASE WHEN rfc3339_shape AND parsed_next_refresh_at IS NOT NULL THEN parsed_next_refresh_at::timestamptz END NULLS FIRST,
id
) AS group_rank
FROM due
SELECT id, api_key, group_last_used_at, due_class, due_at,
row_number() OVER (
PARTITION BY api_key
ORDER BY due_class,
due_at NULLS FIRST,
id
) AS group_rank
FROM candidates
WHERE due_class IS NOT NULL
)
SELECT id
SELECT id, group_last_used_at
FROM ranked
WHERE group_rank = 1
ORDER BY due_class,
CASE WHEN rfc3339_shape AND parsed_next_refresh_at IS NOT NULL THEN parsed_next_refresh_at::timestamptz END NULLS FIRST,
id
LIMIT $2
`, now.UTC(), limit)
ORDER BY due_class, due_at NULLS FIRST, id
LIMIT $4
`, now.UTC(), debounceSeconds, maxWaitSeconds, limit)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
type dueRow struct {
id int64
groupLastUsed *time.Time
}
rowsOut := make([]dueRow, 0, limit)
ids := make([]int64, 0, limit)
for rows.Next() {
var id int64
if err := rows.Scan(&id); err != nil {
var row dueRow
if err := rows.Scan(&row.id, &row.groupLastUsed); err != nil {
return nil, err
}
ids = append(ids, id)
rowsOut = append(rowsOut, row)
ids = append(ids, row.id)
}
if err := rows.Err(); err != nil {
return nil, err
@@ -430,11 +516,26 @@ func (r *accountRepository) ListDueOllamaCloudUsageAccounts(ctx context.Context,
if err != nil {
return nil, err
}
result := make([]service.Account, 0, len(accounts))
byID := make(map[int64]*service.Account, len(accounts))
for _, account := range accounts {
if account != nil {
result = append(result, *account)
byID[account.ID] = account
}
}
result := make([]service.Account, 0, len(rowsOut))
for _, row := range rowsOut {
account := byID[row.id]
if account == nil {
continue
}
// Stamp group MAX(last_used_at) for service due evaluation.
if row.groupLastUsed != nil {
ts := row.groupLastUsed.UTC()
account.LastUsedAt = &ts
} else {
account.LastUsedAt = nil
}
result = append(result, *account)
}
return result, nil
}
@@ -23,81 +23,124 @@ func TestListDueOllamaCloudUsageAccountsOrderingLimitAndProxyHydration(t *testin
Username: "user", Password: "pass", Status: service.StatusActive,
})
createAccount := func(name, baseURL string, proxyID *int64, nextRefreshAt *time.Time) *service.Account {
createAccount := func(name, baseURL string, proxyID *int64, snapshot map[string]any, lastUsed *time.Time) *service.Account {
t.Helper()
extra := map[string]any{
service.OllamaCloudUsageSessionExtraKey: "cipher:wos-session=fixture",
service.OllamaCloudUsageAutoRefreshExtraKey: true,
}
if nextRefreshAt != nil {
extra[service.OllamaCloudUsageSnapshotExtraKey] = map[string]any{
"status": service.OllamaCloudUsageStatusOK, "next_refresh_at": nextRefreshAt.UTC().Format(time.RFC3339Nano),
}
if snapshot != nil {
extra[service.OllamaCloudUsageSnapshotExtraKey] = snapshot
}
return mustCreateAccount(t, tx.Client(), &service.Account{
Name: name, Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": name, "base_url": baseURL},
Extra: extra, ProxyID: proxyID,
Extra: extra, ProxyID: proxyID, LastUsedAt: lastUsed,
})
}
uppercasePath := createAccount("ollama-uppercase-path", "https://ollama.com/V1", nil, nil)
missingSnapshot := createAccount("ollama-due-missing", "HTTPS://WWW.OLLAMA.COM:443/v1", &proxy.ID, nil)
oldest := now.Add(-2 * time.Hour)
due := createAccount("ollama-due-oldest", "https://ollama.com", nil, &oldest)
future := now.Add(time.Minute)
_ = createAccount("ollama-not-due", "https://ollama.com", nil, &future)
_ = createAccount("ollama-ineligible", "https://ollama.com.evil.test", nil, nil)
uppercasePath := createAccount("ollama-uppercase-path", "https://ollama.com/V1", nil, nil, nil)
missingSnapshot := createAccount("ollama-due-missing", "HTTPS://WWW.OLLAMA.COM:443/v1", &proxy.ID, nil, nil)
fetched := now.Add(-2 * time.Hour)
activity := now.Add(-5 * time.Minute)
due := createAccount("ollama-due-activity", "https://ollama.com", nil, map[string]any{
"status": service.OllamaCloudUsageStatusOK,
"fetched_at": fetched.UTC().Format(time.RFC3339Nano),
"last_attempt_at": fetched.UTC().Format(time.RFC3339Nano),
"next_refresh_at": fetched.Add(time.Hour).UTC().Format(time.RFC3339Nano),
}, &activity)
// Success snapshot without newer activity must not be listed.
_ = createAccount("ollama-not-due-idle", "https://ollama.com", nil, map[string]any{
"status": service.OllamaCloudUsageStatusOK,
"fetched_at": now.Add(-time.Hour).UTC().Format(time.RFC3339Nano),
"last_attempt_at": now.Add(-time.Hour).UTC().Format(time.RFC3339Nano),
"next_refresh_at": now.Add(-time.Minute).UTC().Format(time.RFC3339Nano),
}, nil)
_ = createAccount("ollama-ineligible", "https://ollama.com.evil.test", nil, nil, nil)
accounts, err := repo.ListDueOllamaCloudUsageAccounts(ctx, now, 2)
accounts, err := repo.ListDueOllamaCloudUsageAccounts(ctx, now, time.Minute, time.Hour, 2)
require.NoError(t, err)
require.Len(t, accounts, 2)
require.Equal(t, missingSnapshot.ID, accounts[0].ID)
require.Equal(t, due.ID, accounts[1].ID)
require.NotNil(t, accounts[1].LastUsedAt)
require.WithinDuration(t, activity.UTC(), accounts[1].LastUsedAt.UTC(), time.Second)
require.NotContains(t, accountIDs(accounts), uppercasePath.ID)
require.NotNil(t, accounts[0].Proxy)
require.Equal(t, proxy.ID, accounts[0].Proxy.ID)
require.Equal(t, proxy.URL(), accounts[0].Proxy.URL())
}
func TestListDueOllamaCloudUsageAccountsParsesRFC3339NanoAndFailsOpen(t *testing.T) {
func TestListDueOllamaCloudUsageAccountsUsesGroupMaxLastUsedAndFailsOpen(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
repo := newAccountRepositoryWithSQL(tx.Client(), tx, nil)
now := time.Date(2026, time.July, 22, 14, 0, 0, 0, time.UTC)
fetched := now.Add(-30 * time.Minute)
older := now.Add(-10 * time.Minute)
newer := now.Add(-2 * time.Minute)
create := func(name, nextRefreshAt string) *service.Account {
t.Helper()
return mustCreateAccount(t, tx.Client(), &service.Account{
Name: name, Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": name, "base_url": "https://ollama.com"},
Extra: map[string]any{
service.OllamaCloudUsageSessionExtraKey: "cipher:wos-session=fixture",
service.OllamaCloudUsageAutoRefreshExtraKey: true,
service.OllamaCloudUsageSnapshotExtraKey: map[string]any{
"status": service.OllamaCloudUsageStatusOK, "next_refresh_at": nextRefreshAt,
},
leader := mustCreateAccount(t, tx.Client(), &service.Account{
Name: "ollama-group-leader", Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "shared-key", "base_url": "https://ollama.com"},
Extra: map[string]any{
service.OllamaCloudUsageSessionExtraKey: "cipher:wos-session=fixture",
service.OllamaCloudUsageAutoRefreshExtraKey: true,
service.OllamaCloudUsageSnapshotExtraKey: map[string]any{
"status": service.OllamaCloudUsageStatusOK,
"fetched_at": fetched.UTC().Format(time.RFC3339Nano),
"last_attempt_at": fetched.UTC().Format(time.RFC3339Nano),
"next_refresh_at": fetched.Add(time.Hour).UTC().Format(time.RFC3339Nano),
},
})
}
},
LastUsedAt: &older,
})
_ = mustCreateAccount(t, tx.Client(), &service.Account{
Name: "ollama-group-sibling", Platform: service.PlatformAnthropic, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "shared-key", "base_url": "https://www.ollama.com/v1"},
LastUsedAt: &newer,
})
invalid := mustCreateAccount(t, tx.Client(), &service.Account{
Name: "ollama-invalid-snapshot", Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "invalid-key", "base_url": "https://ollama.com"},
Extra: map[string]any{
service.OllamaCloudUsageSessionExtraKey: "cipher:wos-session=fixture",
service.OllamaCloudUsageAutoRefreshExtraKey: true,
service.OllamaCloudUsageSnapshotExtraKey: map[string]any{
"status": service.OllamaCloudUsageStatusOK, "fetched_at": "2026-02-30T09:00:00.123456789Z",
},
},
})
idle := mustCreateAccount(t, tx.Client(), &service.Account{
Name: "ollama-idle-ok", Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "idle-key", "base_url": "https://ollama.com"},
Extra: map[string]any{
service.OllamaCloudUsageSessionExtraKey: "cipher:wos-session=fixture",
service.OllamaCloudUsageAutoRefreshExtraKey: true,
service.OllamaCloudUsageSnapshotExtraKey: map[string]any{
"status": service.OllamaCloudUsageStatusOK,
"fetched_at": fetched.UTC().Format(time.RFC3339Nano),
"last_attempt_at": fetched.UTC().Format(time.RFC3339Nano),
"next_refresh_at": fetched.Add(time.Hour).UTC().Format(time.RFC3339Nano),
},
},
})
sevenDigitsOffset := create("ollama-nano-seven", "2026-07-22T11:00:00.1234567-02:00")
eightDigitsOffset := create("ollama-nano-eight", "2026-07-22T11:00:00.12345678+01:00")
nineDigitsZ := create("ollama-nano-nine", "2026-07-22T09:00:00.123456789Z")
invalidCalendar := create("ollama-nano-invalid", "2026-02-30T09:00:00.123456789Z")
future := create("ollama-nano-future", "2026-07-22T15:00:00.123456789Z")
accounts, err := repo.ListDueOllamaCloudUsageAccounts(ctx, now, 10)
accounts, err := repo.ListDueOllamaCloudUsageAccounts(ctx, now, time.Minute, time.Hour, 10)
require.NoError(t, err, "invalid stored values must not abort the query")
require.Equal(t, []int64{
invalidCalendar.ID,
nineDigitsZ.ID,
eightDigitsOffset.ID,
sevenDigitsOffset.ID,
}, accountIDs(accounts))
require.NotContains(t, accountIDs(accounts), future.ID)
ids := accountIDs(accounts)
require.Contains(t, ids, invalid.ID)
require.Contains(t, ids, leader.ID)
require.NotContains(t, ids, idle.ID)
for _, account := range accounts {
if account.ID == leader.ID {
require.NotNil(t, account.LastUsedAt)
require.WithinDuration(t, newer.UTC(), account.LastUsedAt.UTC(), time.Second,
"group MAX(last_used_at) must come from the sibling")
}
}
}
func TestLockAndMergeAccountProbeExtraCoalescesNullableOllamaGroupIdentity(t *testing.T) {
@@ -413,3 +456,111 @@ func TestUpdateCredentialsUnchangedCredentialsPreserveManagedExtra(t *testing.T)
require.NotContains(t, probeLoaded.Extra, service.UpstreamBillingProbeExtraKey,
"changed credentials must keep clearing the probe snapshot")
}
// TestListDueOllamaCloudUsageAccountsSQLDueRulesMatchService proves the SQL
// candidate layer applies debounce / max-wait / failure-backoff before LIMIT,
// matching service.ollamaCloudUsageIsAutoRefreshDue, and that >20 active-but-
// not-yet-due groups cannot starve a truly due max-wait group.
func TestListDueOllamaCloudUsageAccountsSQLDueRulesMatchService(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
repo := newAccountRepositoryWithSQL(tx.Client(), tx, nil)
now := time.Date(2026, time.July, 25, 12, 0, 0, 0, time.UTC)
debounce := time.Minute
maxWait := time.Hour
createOK := func(name string, fetched, lastUsed time.Time) *service.Account {
t.Helper()
return mustCreateAccount(t, tx.Client(), &service.Account{
Name: name, Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": name, "base_url": "https://ollama.com"},
Extra: map[string]any{
service.OllamaCloudUsageSessionExtraKey: "cipher:wos-session=fixture",
service.OllamaCloudUsageAutoRefreshExtraKey: true,
service.OllamaCloudUsageSnapshotExtraKey: map[string]any{
"status": service.OllamaCloudUsageStatusOK,
"fetched_at": fetched.UTC().Format(time.RFC3339Nano),
"last_attempt_at": fetched.UTC().Format(time.RFC3339Nano),
"next_refresh_at": fetched.Add(maxWait).UTC().Format(time.RFC3339Nano),
},
},
LastUsedAt: &lastUsed,
})
}
createFailed := func(name string, lastAttempt, lastUsed, nextRefresh time.Time, nextRefreshRaw string) *service.Account {
t.Helper()
snapshot := map[string]any{
"status": service.OllamaCloudUsageStatusFailed,
"last_attempt_at": lastAttempt.UTC().Format(time.RFC3339Nano),
"failure_count": 1,
}
if nextRefreshRaw != "" {
snapshot["next_refresh_at"] = nextRefreshRaw
} else {
snapshot["next_refresh_at"] = nextRefresh.UTC().Format(time.RFC3339Nano)
}
return mustCreateAccount(t, tx.Client(), &service.Account{
Name: name, Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": name, "base_url": "https://ollama.com"},
Extra: map[string]any{
service.OllamaCloudUsageSessionExtraKey: "cipher:wos-session=fixture",
service.OllamaCloudUsageAutoRefreshExtraKey: true,
service.OllamaCloudUsageSnapshotExtraKey: snapshot,
},
LastUsedAt: &lastUsed,
})
}
// 21 groups with activity after fetch but debounce not elapsed — previously
// these alone could fill LIMIT 20 every minute and starve true due groups.
notDueIDs := make(map[int64]struct{}, 21)
for i := 0; i < 21; i++ {
// fetched 10m ago, last used 10s ago → due_at = lastUsed+debounce = now+50s (not due)
acc := createOK(fmt.Sprintf("ollama-not-due-debounce-%02d", i), now.Add(-10*time.Minute), now.Add(-10*time.Second))
notDueIDs[acc.ID] = struct{}{}
}
// Truly due via max-wait: fetched 2h ago, continuous activity 10s ago.
// due_at = min(now-10s+1m, now-2h+1h) = now-1h → due.
maxWaitDue := createOK("ollama-due-maxwait", now.Add(-2*time.Hour), now.Add(-10*time.Second))
// Success debounce elapsed: last used 2m ago with debounce 1m → due.
debounceDue := createOK("ollama-due-debounce", now.Add(-30*time.Minute), now.Add(-2*time.Minute))
// Success still within debounce → not due.
_ = createOK("ollama-not-due-fresh", now.Add(-30*time.Minute), now.Add(-20*time.Second))
// Failure blocked by next_refresh_at backoff even with new activity.
_ = createFailed("ollama-fail-backoff", now.Add(-30*time.Minute), now.Add(-2*time.Minute), now.Add(10*time.Minute), "")
// Failure after backoff with new request → due.
failDue := createFailed("ollama-fail-due", now.Add(-30*time.Minute), now.Add(-2*time.Minute), now.Add(-time.Minute), "")
// Invalid next_refresh_at must fail open (not abort query / not block activity due).
failInvalidNext := createFailed("ollama-fail-invalid-next", now.Add(-30*time.Minute), now.Add(-2*time.Minute), time.Time{}, "not-a-timestamp")
accounts, err := repo.ListDueOllamaCloudUsageAccounts(ctx, now, debounce, maxWait, 20)
require.NoError(t, err)
ids := accountIDs(accounts)
require.Contains(t, ids, maxWaitDue.ID, "max-wait due group must not be starved by not-yet-due activity groups")
require.Contains(t, ids, debounceDue.ID, "success debounce elapsed must be due in SQL")
require.Contains(t, ids, failDue.ID, "failure after backoff with new activity must be due in SQL")
require.Contains(t, ids, failInvalidNext.ID, "invalid next_refresh_at must fail open to activity due")
require.LessOrEqual(t, len(accounts), 20)
// Fixtures below match service.ollamaCloudUsageIsAutoRefreshDue semantics;
// none of the not-yet-due groups may appear even when they outnumber the limit.
for _, id := range ids {
_, isNotDue := notDueIDs[id]
require.False(t, isNotDue, "not-yet-due debounce group %d must not be returned by SQL LIMIT layer", id)
}
require.NotContains(t, ids, int64(0))
// Explicit not-due names must stay out: fresh success and failure still in backoff.
for _, account := range accounts {
require.NotContains(t, account.Name, "not-due")
require.NotEqual(t, "ollama-fail-backoff", account.Name)
require.NotEqual(t, "ollama-not-due-fresh", account.Name)
}
}
@@ -190,13 +190,15 @@ func TestListDueOllamaCloudUsageAccountsFiltersOrdersAndLimits(t *testing.T) {
require.NoError(t, err)
t.Cleanup(func() { _ = db.Close() })
now := time.Date(2026, time.July, 22, 12, 0, 0, 0, time.UTC)
debounce := time.Minute
maxWait := time.Hour
var capturedSQL string
mock.ExpectQuery("WITH candidates AS").
WithArgs(now, 20).
WillReturnRows(sqlmock.NewRows([]string{"id"}))
mock.ExpectQuery("WITH eligible AS").
WithArgs(now.UTC(), debounce.Seconds(), maxWait.Seconds(), 20).
WillReturnRows(sqlmock.NewRows([]string{"id", "group_last_used_at"}))
repo := newAccountRepositoryWithSQL(nil, captureQuerySQL{db: db, captured: &capturedSQL}, nil)
accounts, err := repo.ListDueOllamaCloudUsageAccounts(context.Background(), now, 20)
accounts, err := repo.ListDueOllamaCloudUsageAccounts(context.Background(), now, debounce, maxWait, 20)
require.NoError(t, err)
require.Empty(t, accounts)
@@ -209,10 +211,17 @@ func TestListDueOllamaCloudUsageAccountsFiltersOrdersAndLimits(t *testing.T) {
ollamaCloudBaseURLMatchesSQL("credentials ->> 'base_url'"),
"jsonb_typeof(extra -> 'ollama_cloud_usage_session') = 'string'",
`extra @> '{"ollama_cloud_usage_auto_refresh": true}'::jsonb`,
"parsed_next_refresh_at::timestamptz <= $1",
"MAX(last_used_at) AS group_last_used_at",
"PARTITION BY api_key",
"WHERE group_rank = 1",
"LIMIT $2",
"LIMIT $4",
"make_interval(secs => $2::double precision)",
"make_interval(secs => $3::double precision)",
"group_last_used_at > parsed_fetched_at::timestamptz",
"group_last_used_at > parsed_last_attempt_at::timestamptz",
"$1 >= activity_due_at",
"COALESCE(parsed_next_refresh_at::timestamptz, '-infinity'::timestamptz)",
"ORDER BY due_class, due_at NULLS FIRST, id",
} {
require.Contains(t, normalized, clause)
}
@@ -1623,3 +1623,114 @@ func TestGatewayService_AnthropicAPIKeyPassthrough_StreamingUpstreamReadErrorAft
require.True(t, result.clientDisconnect)
require.Equal(t, 8, result.usage.InputTokens)
}
func TestGatewayService_AnthropicAPIKeyPassthrough_TransportErrorRecordsOllamaActivity(t *testing.T) {
gin.SetMode(gin.TestMode)
deferred := NewDeferredService(nil, nil, time.Second)
upstream := &anthropicHTTPUpstreamRecorder{err: errors.New("dial tcp timeout")}
svc := &GatewayService{
cfg: &config.Config{
Security: config.SecurityConfig{
URLAllowlist: config.URLAllowlistConfig{Enabled: false},
},
},
httpUpstream: upstream,
deferredService: deferred,
}
ollama := &Account{
ID: 601, Name: "ollama-anthropic", Platform: PlatformAnthropic, Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{"api_key": "k-ollama", "base_url": "https://ollama.com"},
Extra: map[string]any{"anthropic_passthrough": true},
Status: StatusActive, Schedulable: true,
}
other := newAnthropicAPIKeyAccountForTest()
other.ID = 602
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil)
_, err := svc.forwardAnthropicAPIKeyPassthrough(context.Background(), c, ollama, []byte(`{"model":"x"}`), "x", "x", false, time.Now())
require.Error(t, err)
rec2 := httptest.NewRecorder()
c2, _ := gin.CreateTestContext(rec2)
c2.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil)
_, err = svc.forwardAnthropicAPIKeyPassthrough(context.Background(), c2, other, []byte(`{"model":"x"}`), "x", "x", false, time.Now())
require.Error(t, err)
_, ok := deferred.lastUsedUpdates.Load(int64(601))
require.True(t, ok, "Anthropic passthrough transport error on Ollama account must record activity")
_, ok = deferred.lastUsedUpdates.Load(int64(602))
require.False(t, ok, "non-Ollama Anthropic passthrough transport error must not record Ollama activity")
}
func TestGatewayService_AnthropicAPIKeyPassthrough_ContextCanceledSkipsOllamaActivity(t *testing.T) {
gin.SetMode(gin.TestMode)
deferred := NewDeferredService(nil, nil, time.Second)
upstream := &anthropicHTTPUpstreamRecorder{err: context.Canceled}
svc := &GatewayService{
cfg: &config.Config{
Security: config.SecurityConfig{
URLAllowlist: config.URLAllowlistConfig{Enabled: false},
},
},
httpUpstream: upstream,
deferredService: deferred,
}
ollama := &Account{
ID: 603, Name: "ollama-canceled", Platform: PlatformAnthropic, Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{"api_key": "k-ollama", "base_url": "https://ollama.com"},
Extra: map[string]any{"anthropic_passthrough": true},
Status: StatusActive, Schedulable: true,
}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil)
_, err := svc.forwardAnthropicAPIKeyPassthrough(context.Background(), c, ollama, []byte(`{"model":"x"}`), "x", "x", false, time.Now())
require.Error(t, err)
_, ok := deferred.lastUsedUpdates.Load(int64(603))
require.False(t, ok, "context.Canceled on Anthropic passthrough must not count as Ollama activity")
}
func TestGatewayService_AnthropicAPIKeyPassthrough_Non2xxRecordsOllamaActivity(t *testing.T) {
gin.SetMode(gin.TestMode)
deferred := NewDeferredService(nil, nil, time.Second)
// 400 is non-retryable / non-failover for default API-key accounts, so it reaches handleErrorResponse.
upstream := &anthropicHTTPUpstreamRecorder{
resp: &http.Response{
StatusCode: http.StatusBadRequest,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(`{"type":"error","error":{"type":"invalid_request_error","message":"bad"}}`)),
},
}
svc := &GatewayService{
cfg: &config.Config{
Security: config.SecurityConfig{
URLAllowlist: config.URLAllowlistConfig{Enabled: false},
},
},
httpUpstream: upstream,
deferredService: deferred,
rateLimitService: &RateLimitService{},
}
ollama := &Account{
ID: 604, Name: "ollama-400", Platform: PlatformAnthropic, Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{"api_key": "k-ollama", "base_url": "https://ollama.com"},
Extra: map[string]any{"anthropic_passthrough": true},
Status: StatusActive, Schedulable: true,
}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil)
_, _ = svc.forwardAnthropicAPIKeyPassthrough(context.Background(), c, ollama, []byte(`{"model":"x"}`), "x", "x", false, time.Now())
_, ok := deferred.lastUsedUpdates.Load(int64(604))
require.True(t, ok, "Anthropic passthrough non-2xx on Ollama account must record activity via handleErrorResponse")
}
@@ -114,6 +114,9 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
if resp != nil && resp.Body != nil {
_ = resp.Body.Close()
}
if !errors.Is(err, context.Canceled) {
scheduleOllamaCloudUsageActivity(s.deferredService, account)
}
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
@@ -369,6 +369,10 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
if resp != nil && resp.Body != nil {
_ = resp.Body.Close()
}
// Transport attempt left local validation; count Ollama Cloud activity.
if !errors.Is(err, context.Canceled) {
scheduleOllamaCloudUsageActivity(s.deferredService, account)
}
// Ensure the client receives an error response (handlers assume Forward writes on non-failover errors).
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
@@ -356,6 +356,8 @@ func (s *GatewayService) readUpstreamErrorBody(resp *http.Response) ([]byte, err
}
func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Response, c *gin.Context, account *Account, requestedModel ...string) (*ForwardResult, error) {
// Upstream returned a non-success HTTP status; count Ollama Cloud activity.
scheduleOllamaCloudUsageActivity(s.deferredService, account)
body, readErr := s.readUpstreamErrorBody(resp)
if readErr != nil {
// 读取失败时 body 可能被截断,错误分类会基于不完整数据;记录日志以便排查,
+165 -10
View File
@@ -36,6 +36,9 @@ const (
ollamaCloudUsageDefaultIntervalMinutes = 60
ollamaCloudUsageMinIntervalMinutes = 15
ollamaCloudUsageMaxIntervalMinutes = 24 * 60
ollamaCloudUsageDefaultDebounceMinutes = 1
ollamaCloudUsageMinDebounceMinutes = 1
ollamaCloudUsageMaxDebounceMinutes = 60
ollamaCloudUsageCycleInterval = time.Minute
ollamaCloudUsageManualRefreshInterval = 30 * time.Second
ollamaCloudUsageRequestTimeout = 15 * time.Second
@@ -76,10 +79,15 @@ const (
OllamaCloudUsageStatusFailed = "failed"
)
// OllamaCloudUsageSettings controls the opt-in periodic refresh runner.
// OllamaCloudUsageSettings controls the opt-in request-driven refresh runner.
//
// IntervalMinutes is the max-wait bound: when model requests keep arriving and
// the trailing debounce keeps sliding, a refresh is forced after this long.
// DebounceMinutes is the quiet period after the latest request in a group.
type OllamaCloudUsageSettings struct {
Enabled bool `json:"enabled"`
IntervalMinutes int `json:"interval_minutes"`
IntervalMinutes int `json:"interval_minutes"` // max wait while requests continue
DebounceMinutes int `json:"debounce_minutes"` // trailing quiet period after last request
}
// OllamaCloudUsageWindow is a narrow, sanitized view of one official usage window.
@@ -114,6 +122,12 @@ type OllamaCloudUsageData struct {
}
// OllamaCloudUsageSnapshot is the only usage observation persisted in account extra.
//
// NextRefreshAt remains a persisted compatibility field. For status=ok it is a
// max-wait horizon marker only; automatic success refreshes are driven by model
// request activity (group last_used_at + debounce/max-wait), not by this field
// alone. For failed/unauthorized snapshots it is the failure not-before time
// (Retry-After / exponential backoff) and is enforced as max(activityDue, NextRefreshAt).
type OllamaCloudUsageSnapshot struct {
Status string `json:"status"`
Data *OllamaCloudUsageData `json:"data,omitempty"`
@@ -142,7 +156,7 @@ type ollamaCloudUsageRepository interface {
SetOllamaCloudUsageAutoRefresh(context.Context, *Account, bool) error
UpdateOllamaCloudUsageSnapshot(context.Context, *Account, *OllamaCloudUsageSnapshot) error
DisableOllamaCloudUsageAutoRefresh(context.Context, *Account) error
ListDueOllamaCloudUsageAccounts(context.Context, time.Time, int) ([]Account, error)
ListDueOllamaCloudUsageAccounts(context.Context, time.Time, time.Duration, time.Duration, int) ([]Account, error)
}
// GetOllamaCloudUsageSettings returns fail-safe defaults when the setting is absent.
@@ -168,6 +182,9 @@ func (s *SettingService) GetOllamaCloudUsageSettings(ctx context.Context) (*Olla
if settings.IntervalMinutes == 0 {
settings.IntervalMinutes = defaults.IntervalMinutes
}
if settings.DebounceMinutes == 0 {
settings.DebounceMinutes = defaults.DebounceMinutes
}
normalizeOllamaCloudUsageSettings(&settings)
return &settings, nil
}
@@ -179,12 +196,22 @@ func (s *SettingService) SetOllamaCloudUsageSettings(ctx context.Context, settin
if settings == nil {
return infraerrors.BadRequest("INVALID_OLLAMA_CLOUD_USAGE_SETTINGS", "settings cannot be nil")
}
if settings.DebounceMinutes == 0 {
// Legacy clients that omit debounce_minutes keep the fail-safe default.
settings.DebounceMinutes = ollamaCloudUsageDefaultDebounceMinutes
}
if settings.IntervalMinutes < ollamaCloudUsageMinIntervalMinutes || settings.IntervalMinutes > ollamaCloudUsageMaxIntervalMinutes {
return infraerrors.BadRequest(
"INVALID_OLLAMA_CLOUD_USAGE_INTERVAL",
fmt.Sprintf("interval_minutes must be between %d and %d", ollamaCloudUsageMinIntervalMinutes, ollamaCloudUsageMaxIntervalMinutes),
)
}
if settings.DebounceMinutes < ollamaCloudUsageMinDebounceMinutes || settings.DebounceMinutes > ollamaCloudUsageMaxDebounceMinutes {
return infraerrors.BadRequest(
"INVALID_OLLAMA_CLOUD_USAGE_DEBOUNCE",
fmt.Sprintf("debounce_minutes must be between %d and %d", ollamaCloudUsageMinDebounceMinutes, ollamaCloudUsageMaxDebounceMinutes),
)
}
normalizeOllamaCloudUsageSettings(settings)
data, err := json.Marshal(settings)
if err != nil {
@@ -194,7 +221,11 @@ func (s *SettingService) SetOllamaCloudUsageSettings(ctx context.Context, settin
}
func defaultOllamaCloudUsageSettings() *OllamaCloudUsageSettings {
return &OllamaCloudUsageSettings{Enabled: false, IntervalMinutes: ollamaCloudUsageDefaultIntervalMinutes}
return &OllamaCloudUsageSettings{
Enabled: false,
IntervalMinutes: ollamaCloudUsageDefaultIntervalMinutes,
DebounceMinutes: ollamaCloudUsageDefaultDebounceMinutes,
}
}
func normalizeOllamaCloudUsageSettings(settings *OllamaCloudUsageSettings) {
@@ -204,6 +235,116 @@ func normalizeOllamaCloudUsageSettings(settings *OllamaCloudUsageSettings) {
if settings.IntervalMinutes > ollamaCloudUsageMaxIntervalMinutes {
settings.IntervalMinutes = ollamaCloudUsageMaxIntervalMinutes
}
if settings.DebounceMinutes <= 0 {
settings.DebounceMinutes = ollamaCloudUsageDefaultDebounceMinutes
}
if settings.DebounceMinutes < ollamaCloudUsageMinDebounceMinutes {
settings.DebounceMinutes = ollamaCloudUsageMinDebounceMinutes
}
if settings.DebounceMinutes > ollamaCloudUsageMaxDebounceMinutes {
settings.DebounceMinutes = ollamaCloudUsageMaxDebounceMinutes
}
}
func ollamaCloudUsageDurations(settings *OllamaCloudUsageSettings) (debounce, maxWait time.Duration) {
normalized := defaultOllamaCloudUsageSettings()
if settings != nil {
*normalized = *settings
}
normalizeOllamaCloudUsageSettings(normalized)
return time.Duration(normalized.DebounceMinutes) * time.Minute,
time.Duration(normalized.IntervalMinutes) * time.Minute
}
// ollamaCloudUsageIsAutoRefreshDue decides whether a configured auto-refresh
// group should fetch now. groupLastUsedAt must be MAX(last_used_at) across the
// exact api_key group so shared multi-platform accounts do not miss activity.
//
// Success: a request must be newer than fetched_at; dueAt = min(lastUsed+debounce, fetchedAt+maxWait).
// Failure: a request must be newer than last_attempt_at; activity due uses the same min formula,
// then dueAt = max(activityDue, next_refresh_at) so Retry-After / exponential backoff win.
// Missing or invalid snapshots fail open to a first fetch.
func ollamaCloudUsageIsAutoRefreshDue(
snapshot *OllamaCloudUsageSnapshot,
groupLastUsedAt *time.Time,
now time.Time,
debounce, maxWait time.Duration,
) bool {
dueAt, ok := ollamaCloudUsageAutoRefreshDueAt(snapshot, groupLastUsedAt, debounce, maxWait)
if !ok {
return false
}
return !now.Before(dueAt)
}
func ollamaCloudUsageAutoRefreshDueAt(
snapshot *OllamaCloudUsageSnapshot,
groupLastUsedAt *time.Time,
debounce, maxWait time.Duration,
) (time.Time, bool) {
if debounce <= 0 {
debounce = time.Duration(ollamaCloudUsageDefaultDebounceMinutes) * time.Minute
}
if maxWait <= 0 {
maxWait = time.Duration(ollamaCloudUsageDefaultIntervalMinutes) * time.Minute
}
if snapshot == nil {
return time.Time{}, true
}
switch snapshot.Status {
case OllamaCloudUsageStatusOK:
if snapshot.FetchedAt == nil || snapshot.FetchedAt.IsZero() {
return time.Time{}, true
}
fetchedAt := snapshot.FetchedAt.UTC()
if groupLastUsedAt == nil || !groupLastUsedAt.After(fetchedAt) {
return time.Time{}, false
}
lastUsed := groupLastUsedAt.UTC()
return minTime(lastUsed.Add(debounce), fetchedAt.Add(maxWait)), true
case OllamaCloudUsageStatusFailed, OllamaCloudUsageStatusUnauthorized:
if snapshot.LastAttemptAt.IsZero() {
return time.Time{}, true
}
lastAttempt := snapshot.LastAttemptAt.UTC()
if groupLastUsedAt == nil || !groupLastUsedAt.After(lastAttempt) {
return time.Time{}, false
}
lastUsed := groupLastUsedAt.UTC()
activityDue := minTime(lastUsed.Add(debounce), lastAttempt.Add(maxWait))
if !snapshot.NextRefreshAt.IsZero() && snapshot.NextRefreshAt.UTC().After(activityDue) {
return snapshot.NextRefreshAt.UTC(), true
}
return activityDue, true
default:
return time.Time{}, true
}
}
// maxOllamaCloudUsageGroupLastUsed returns the newest last_used_at among group members.
func maxOllamaCloudUsageGroupLastUsed(accounts []Account) *time.Time {
var latest *time.Time
for i := range accounts {
candidate := accounts[i].LastUsedAt
if candidate == nil || candidate.IsZero() {
continue
}
if latest == nil || candidate.After(*latest) {
ts := candidate.UTC()
latest = &ts
}
}
return latest
}
// scheduleOllamaCloudUsageActivity records that an Ollama Cloud API-key account
// actually attempted an upstream model request (including 429/5xx/transport errors).
// Local auth/validation failures must not call this. DeferredService dedupes writes.
func scheduleOllamaCloudUsageActivity(deferred *DeferredService, account *Account) {
if deferred == nil || account == nil || !IsOllamaCloudUsageAccount(account) {
return
}
deferred.ScheduleLastUsedUpdate(account.ID)
}
// OllamaCloudUsageService refreshes the official settings HTML without affecting routing state.
@@ -531,7 +672,7 @@ func (s *OllamaCloudUsageService) Refresh(ctx context.Context, accountID int64)
if err != nil {
return nil, err
}
if _, err := s.refreshAccount(ctx, accountID, settings.IntervalMinutes, false); err != nil {
if _, err := s.refreshAccount(ctx, accountID, settings, false); err != nil {
return nil, err
}
return s.GetState(ctx, accountID)
@@ -561,7 +702,8 @@ func (s *OllamaCloudUsageService) RunDue(ctx context.Context) error {
return ErrOllamaCloudUsageUnavailable
}
now := s.currentTime()
accounts, err := writer.ListDueOllamaCloudUsageAccounts(ctx, now, ollamaCloudUsageMaxPerCycle)
debounce, maxWait := ollamaCloudUsageDurations(settings)
accounts, err := writer.ListDueOllamaCloudUsageAccounts(ctx, now, debounce, maxWait, ollamaCloudUsageMaxPerCycle)
if err != nil {
return fmt.Errorf("list due Ollama Cloud usage accounts: %w", err)
}
@@ -577,13 +719,15 @@ func (s *OllamaCloudUsageService) RunDue(ctx context.Context) error {
continue
}
seenGroups[fingerprint] = struct{}{}
if snapshot := decodeOllamaCloudUsageSnapshot(account.Extra); snapshot != nil && now.Before(snapshot.NextRefreshAt) {
snapshot := decodeOllamaCloudUsageSnapshot(account.Extra)
// ListDue stamps Account.LastUsedAt with the api_key group MAX(last_used_at).
if !ollamaCloudUsageIsAutoRefreshDue(snapshot, account.LastUsedAt, now, debounce, maxWait) {
continue
}
accountID := account.ID
expected := account
group.Go(func() error {
if _, refreshErr := s.refreshAccount(ctx, accountID, settings.IntervalMinutes, true); refreshErr != nil {
if _, refreshErr := s.refreshAccount(ctx, accountID, settings, true); refreshErr != nil {
if errors.Is(refreshErr, ErrOllamaCloudUsageIdentityChanged) {
if disableErr := writer.DisableOllamaCloudUsageAutoRefresh(ctx, &expected); disableErr != nil {
logger.LegacyPrintf("service.ollama_cloud_usage", "disable_auto_refresh_failed: account_id=%d err=%v", accountID, disableErr)
@@ -598,10 +742,15 @@ func (s *OllamaCloudUsageService) RunDue(ctx context.Context) error {
return group.Wait()
}
func (s *OllamaCloudUsageService) refreshAccount(ctx context.Context, accountID int64, intervalMinutes int, requireEnabled bool) (*OllamaCloudUsageSnapshot, error) {
func (s *OllamaCloudUsageService) refreshAccount(ctx context.Context, accountID int64, settings *OllamaCloudUsageSettings, requireEnabled bool) (*OllamaCloudUsageSnapshot, error) {
if s == nil || s.accountRepo == nil {
return nil, ErrOllamaCloudUsageUnavailable
}
if settings == nil {
settings = defaultOllamaCloudUsageSettings()
}
intervalMinutes := settings.IntervalMinutes
debounce, maxWait := ollamaCloudUsageDurations(settings)
anchor, err := s.accountRepo.GetByID(ctx, accountID)
if err != nil {
return nil, err
@@ -650,7 +799,13 @@ func (s *OllamaCloudUsageService) refreshAccount(ctx context.Context, accountID
if !account.IsActive() || !ollamaCloudUsageAutoRefreshEnabled(account) {
return nil, nil
}
if snapshot := decodeOllamaCloudUsageSnapshot(account.Extra); snapshot != nil && s.currentTime().Before(snapshot.NextRefreshAt) {
groupLastUsed := account.LastUsedAt
if writer, ok := s.accountRepo.(ollamaCloudUsageRepository); ok {
if siblings, listErr := writer.ListOllamaCloudUsageGroupAccounts(ctx, []*Account{account}); listErr == nil {
groupLastUsed = maxOllamaCloudUsageGroupLastUsed(siblings)
}
}
if !ollamaCloudUsageIsAutoRefreshDue(decodeOllamaCloudUsageSnapshot(account.Extra), groupLastUsed, s.currentTime(), debounce, maxWait) {
return nil, nil
}
}
@@ -185,7 +185,7 @@ func applyOllamaUsageTestManagedExtra(account, source *Account) {
}
}
func (r *ollamaUsageTestRepo) ListDueOllamaCloudUsageAccounts(_ context.Context, _ time.Time, limit int) ([]Account, error) {
func (r *ollamaUsageTestRepo) ListDueOllamaCloudUsageAccounts(_ context.Context, _ time.Time, _, _ time.Duration, limit int) ([]Account, error) {
r.mu.Lock()
defer r.mu.Unlock()
if len(r.due) > 0 {
@@ -313,15 +313,91 @@ func TestOllamaCloudUsageSettingsDefaultOffAndValidation(t *testing.T) {
require.NoError(t, err)
require.False(t, settings.Enabled)
require.Equal(t, 60, settings.IntervalMinutes)
require.Equal(t, 1, settings.DebounceMinutes)
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 14})
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 14, DebounceMinutes: 1})
require.Error(t, err)
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 90})
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 90, DebounceMinutes: 61})
require.Error(t, err)
// DebounceMinutes=0 (legacy omit) defaults to 1 on write.
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 90, DebounceMinutes: 0})
require.NoError(t, err)
settings, err = settingsService.GetOllamaCloudUsageSettings(context.Background())
require.NoError(t, err)
require.Equal(t, 1, settings.DebounceMinutes)
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 90, DebounceMinutes: 2})
require.NoError(t, err)
settings, err = settingsService.GetOllamaCloudUsageSettings(context.Background())
require.NoError(t, err)
require.True(t, settings.Enabled)
require.Equal(t, 90, settings.IntervalMinutes)
require.Equal(t, 2, settings.DebounceMinutes)
// Legacy JSON without debounce_minutes defaults to 1.
repo.values[SettingKeyOllamaCloudUsageSettings] = `{"enabled":true,"interval_minutes":45}`
settings, err = settingsService.GetOllamaCloudUsageSettings(context.Background())
require.NoError(t, err)
require.Equal(t, 45, settings.IntervalMinutes)
require.Equal(t, 1, settings.DebounceMinutes)
}
func TestOllamaCloudUsageIsAutoRefreshDue(t *testing.T) {
debounce := time.Minute
maxWait := time.Hour
now := time.Date(2026, time.July, 25, 12, 0, 0, 0, time.UTC)
fetched := now.Add(-30 * time.Minute)
ptr := func(ts time.Time) *time.Time { return &ts }
require.True(t, ollamaCloudUsageIsAutoRefreshDue(nil, nil, now, debounce, maxWait), "missing snapshot first due")
require.True(t, ollamaCloudUsageIsAutoRefreshDue(&OllamaCloudUsageSnapshot{Status: "bogus"}, nil, now, debounce, maxWait), "invalid status first due")
okSnap := &OllamaCloudUsageSnapshot{
Status: OllamaCloudUsageStatusOK, FetchedAt: ptr(fetched),
LastAttemptAt: fetched, NextRefreshAt: fetched.Add(maxWait),
}
require.False(t, ollamaCloudUsageIsAutoRefreshDue(okSnap, nil, now, debounce, maxWait), "no request after success")
require.False(t, ollamaCloudUsageIsAutoRefreshDue(okSnap, ptr(fetched), now, debounce, maxWait), "request not after fetched_at")
require.False(t, ollamaCloudUsageIsAutoRefreshDue(okSnap, ptr(now.Add(-30*time.Second)), now, debounce, maxWait), "debounce not elapsed")
require.True(t, ollamaCloudUsageIsAutoRefreshDue(okSnap, ptr(now.Add(-time.Minute)), now, debounce, maxWait), "single request quiet for debounce")
// Continuous requests: last used is now, but max-wait from old fetch forces due.
oldFetched := now.Add(-2 * time.Hour)
oldSnap := &OllamaCloudUsageSnapshot{
Status: OllamaCloudUsageStatusOK, FetchedAt: ptr(oldFetched),
LastAttemptAt: oldFetched, NextRefreshAt: oldFetched.Add(maxWait),
}
require.True(t, ollamaCloudUsageIsAutoRefreshDue(oldSnap, ptr(now), now, debounce, maxWait), "max-wait forces due while requests continue")
// First request after a very old snapshot is immediately due because fetched+maxWait is past.
require.True(t, ollamaCloudUsageIsAutoRefreshDue(oldSnap, ptr(now.Add(-time.Second)), now, debounce, maxWait), "stale snapshot first request immediate")
failSnap := &OllamaCloudUsageSnapshot{
Status: OllamaCloudUsageStatusFailed, FetchedAt: ptr(fetched),
LastAttemptAt: now.Add(-10 * time.Minute), NextRefreshAt: now.Add(20 * time.Minute),
}
require.False(t, ollamaCloudUsageIsAutoRefreshDue(failSnap, nil, now, debounce, maxWait), "failure without new request")
require.False(t, ollamaCloudUsageIsAutoRefreshDue(failSnap, ptr(now.Add(-time.Minute)), now, debounce, maxWait), "failure blocked by backoff")
failSnap.NextRefreshAt = now.Add(-time.Second)
require.True(t, ollamaCloudUsageIsAutoRefreshDue(failSnap, ptr(now.Add(-time.Minute)), now, debounce, maxWait), "failure after backoff with new request")
require.True(t, ollamaCloudUsageIsAutoRefreshDue(&OllamaCloudUsageSnapshot{
Status: OllamaCloudUsageStatusOK, LastAttemptAt: now,
}, nil, now, debounce, maxWait), "ok without fetched_at fails open")
}
func TestScheduleOllamaCloudUsageActivityOnlyForOllama(t *testing.T) {
deferred := NewDeferredService(nil, nil, time.Second)
ollama := ollamaUsageAccount(1)
other := ollamaUsageAccount(2)
other.Credentials["base_url"] = "https://api.openai.com"
scheduleOllamaCloudUsageActivity(deferred, ollama)
scheduleOllamaCloudUsageActivity(deferred, other)
scheduleOllamaCloudUsageActivity(nil, ollama)
_, ok := deferred.lastUsedUpdates.Load(int64(1))
require.True(t, ok)
_, ok = deferred.lastUsedUpdates.Load(int64(2))
require.False(t, ok)
}
func TestIsOllamaCloudUsageAccountStrictOfficialHost(t *testing.T) {
@@ -51,6 +51,10 @@ func (s *OpenAIGatewayService) handleOpenAIAccountUpstreamError(ctx context.Cont
if account != nil && account.Platform == PlatformGrok && isGrokContentPolicyRejection(statusCode, responseBody) {
return false
}
// Any non-2xx upstream HTTP response means the model request was actually sent.
if s != nil {
scheduleOllamaCloudUsageActivity(s.deferredService, account)
}
stateCtx, cancel := openAIAccountStateContext(ctx)
defer cancel()
@@ -124,6 +124,11 @@ func (s *OpenAIGatewayService) handleOpenAIUpstreamTransportError(ctx context.Co
return err
}
// Transport attempt reached the network path; count as Ollama Cloud activity.
if s != nil {
scheduleOllamaCloudUsageActivity(s.deferredService, account)
}
if classifyOpenAITransportError(err).Persistent {
s.tempUnscheduleOpenAITransportError(ctx, account, safeErr)
}
@@ -210,3 +210,68 @@ func TestForwardAsRawChatCompletions_TransportErrorFailsOver(t *testing.T) {
require.Empty(t, repo.tempUnschedCalls, "plain EOF is transient: fail over but do not evict")
require.Equal(t, 0, rec.Body.Len(), "service must not write a hard 502 before handler can fail over")
}
func TestHandleOpenAIUpstreamTransportError_RecordsOllamaActivityOnly(t *testing.T) {
deferred := NewDeferredService(nil, nil, time.Second)
svc := &OpenAIGatewayService{
accountRepo: &openaiTransportAccountRepoStub{},
deferredService: deferred,
}
ollama := &Account{
ID: 501, Name: "ollama-cloud", Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "k-ollama", "base_url": "https://ollama.com"},
}
other := &Account{
ID: 502, Name: "openai-official", Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "k-openai", "base_url": "https://api.openai.com"},
}
c, _ := newOpenAITransportErrTestContext()
_ = svc.handleOpenAIUpstreamTransportError(context.Background(), c, ollama, errors.New("connection reset"), false)
_ = svc.handleOpenAIUpstreamTransportError(context.Background(), c, other, errors.New("connection reset"), false)
_, ok := deferred.lastUsedUpdates.Load(int64(501))
require.True(t, ok, "Ollama Cloud transport error must schedule last_used activity")
_, ok = deferred.lastUsedUpdates.Load(int64(502))
require.False(t, ok, "non-Ollama transport error must not schedule Ollama activity")
}
func TestHandleOpenAIUpstreamTransportError_ContextCanceledSkipsOllamaActivity(t *testing.T) {
deferred := NewDeferredService(nil, nil, time.Second)
svc := &OpenAIGatewayService{
accountRepo: &openaiTransportAccountRepoStub{},
deferredService: deferred,
}
ollama := &Account{
ID: 503, Name: "ollama-canceled", Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "k-ollama", "base_url": "https://ollama.com"},
}
c, _ := newOpenAITransportErrTestContext()
err := svc.handleOpenAIUpstreamTransportError(context.Background(), c, ollama, context.Canceled, false)
require.ErrorIs(t, err, context.Canceled)
_, ok := deferred.lastUsedUpdates.Load(int64(503))
require.False(t, ok, "context.Canceled is client disconnect before a fault; do not count as Ollama activity")
}
func TestHandleOpenAIAccountUpstreamError_RecordsOllamaActivityOnly(t *testing.T) {
deferred := NewDeferredService(nil, nil, time.Second)
svc := &OpenAIGatewayService{deferredService: deferred}
ollama := &Account{
ID: 504, Name: "ollama-429", Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "k-ollama", "base_url": "https://ollama.com"},
}
other := &Account{
ID: 505, Name: "openai-429", Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "k-openai", "base_url": "https://api.openai.com"},
}
_ = svc.handleOpenAIAccountUpstreamError(context.Background(), ollama, http.StatusTooManyRequests, http.Header{}, []byte(`{"error":{"message":"rate"}}`), "gpt-test")
_ = svc.handleOpenAIAccountUpstreamError(context.Background(), other, http.StatusTooManyRequests, http.Header{}, []byte(`{"error":{"message":"rate"}}`), "gpt-test")
_, ok := deferred.lastUsedUpdates.Load(int64(504))
require.True(t, ok, "Ollama Cloud non-2xx must schedule last_used activity")
_, ok = deferred.lastUsedUpdates.Load(int64(505))
require.False(t, ok, "non-Ollama non-2xx must not schedule Ollama activity")
}
@@ -38,7 +38,7 @@ describe('admin Ollama Cloud usage API', () => {
})
it('uses dedicated global settings endpoints', async () => {
const settings = { enabled: false, interval_minutes: 60 }
const settings = { enabled: false, interval_minutes: 60, debounce_minutes: 1 }
get.mockResolvedValueOnce({ data: settings })
put.mockResolvedValueOnce({ data: settings })
@@ -335,11 +335,13 @@ export default {
},
ollamaCloudUsage: {
title: 'Ollama Cloud Usage Refresh',
description: 'Periodically refresh official Ollama settings-page usage for individually opted-in accounts. Disabled by default.',
description: 'Refresh official Ollama settings-page usage driven by model requests for individually opted-in accounts. Disabled by default. Idle accounts are not polled.',
enabled: 'Enable global automatic refresh',
enabledHint: 'Only accounts with a stored browser session and their own automatic refresh switch enabled are refreshed. Manual refresh remains available.',
intervalMinutes: 'Refresh interval (minutes)',
intervalHint: 'Range: 15–1440 minutes. Failures use bounded exponential backoff.',
enabledHint: 'Only accounts with a stored browser session and their own automatic refresh switch enabled are refreshed, and only after subsequent model requests. Manual refresh remains available.',
intervalMinutes: 'Max wait while requests continue (minutes)',
intervalHint: 'Range: 15–1440 minutes. When continuous requests keep sliding the debounce, force a refresh after this wait.',
debounceMinutes: 'Quiet period after last request (minutes)',
debounceHint: 'Range: 1–60 minutes. Refresh after the latest model request has been quiet for this long.',
saved: 'Ollama Cloud usage refresh settings saved',
saveFailed: 'Failed to save Ollama Cloud usage refresh settings'
},
@@ -328,11 +328,13 @@ export default {
},
ollamaCloudUsage: {
title: 'Ollama Cloud 用量刷新',
description: '定期刷新账号在 Ollama 官方设置页展示的用量;默认关闭。',
description: '在模型请求驱动下刷新账号在 Ollama 官方设置页展示的用量;默认关闭。无新请求时不会自动抓取。',
enabled: '启用全局自动刷新',
enabledHint: '仅刷新已保存浏览器会话且账号自身也开启自动刷新的账号;手动刷新不受影响。',
intervalMinutes: '刷新周期(分钟)',
intervalHint: '范围 15–1440 分钟。失败后按有上限的指数退避重试。',
enabledHint: '仅刷新已保存浏览器会话且账号自身也开启自动刷新的账号;需有后续模型请求才会触发。手动刷新不受影响。',
intervalMinutes: '持续请求最长等待(分钟)',
intervalHint: '范围 15–1440 分钟。请求持续不断导致 debounce 一直后移时,最晚在此时间强制刷新。',
debounceMinutes: '请求安静等待(分钟)',
debounceHint: '范围 1–60 分钟。最后一次模型请求安静满此时长后再抓取用量。',
saved: 'Ollama Cloud 用量刷新设置已保存',
saveFailed: '保存 Ollama Cloud 用量刷新设置失败'
},
+3
View File
@@ -1023,7 +1023,10 @@ export interface OllamaCloudUsageState {
export interface OllamaCloudUsageSettings {
enabled: boolean
/** Max wait while model requests keep arriving (minutes). */
interval_minutes: number
/** Trailing quiet period after the latest model request (minutes). */
debounce_minutes: number
}
export interface Account {
+38 -17
View File
@@ -4220,23 +4220,43 @@
data-testid="ollama-cloud-usage-global-enabled"
/>
</div>
<div v-if="ollamaCloudUsageForm.enabled" class="border-t border-gray-100 pt-4 dark:border-dark-700">
<label class="mb-2 block text-sm font-medium text-gray-700 dark:text-gray-300" for="ollama-cloud-usage-interval">
{{ t("admin.settings.ollamaCloudUsage.intervalMinutes") }}
</label>
<input
id="ollama-cloud-usage-interval"
v-model.number="ollamaCloudUsageForm.interval_minutes"
type="number"
min="15"
max="1440"
class="input w-32"
data-testid="ollama-cloud-usage-global-interval"
@keydown.enter.prevent="saveOllamaCloudUsageSettings"
/>
<p class="mt-1.5 text-xs text-gray-500 dark:text-gray-400">
{{ t("admin.settings.ollamaCloudUsage.intervalHint") }}
</p>
<div v-if="ollamaCloudUsageForm.enabled" class="space-y-4 border-t border-gray-100 pt-4 dark:border-dark-700">
<div>
<label class="mb-2 block text-sm font-medium text-gray-700 dark:text-gray-300" for="ollama-cloud-usage-debounce">
{{ t("admin.settings.ollamaCloudUsage.debounceMinutes") }}
</label>
<input
id="ollama-cloud-usage-debounce"
v-model.number="ollamaCloudUsageForm.debounce_minutes"
type="number"
min="1"
max="60"
class="input w-32"
data-testid="ollama-cloud-usage-global-debounce"
@keydown.enter.prevent="saveOllamaCloudUsageSettings"
/>
<p class="mt-1.5 text-xs text-gray-500 dark:text-gray-400">
{{ t("admin.settings.ollamaCloudUsage.debounceHint") }}
</p>
</div>
<div>
<label class="mb-2 block text-sm font-medium text-gray-700 dark:text-gray-300" for="ollama-cloud-usage-interval">
{{ t("admin.settings.ollamaCloudUsage.intervalMinutes") }}
</label>
<input
id="ollama-cloud-usage-interval"
v-model.number="ollamaCloudUsageForm.interval_minutes"
type="number"
min="15"
max="1440"
class="input w-32"
data-testid="ollama-cloud-usage-global-interval"
@keydown.enter.prevent="saveOllamaCloudUsageSettings"
/>
<p class="mt-1.5 text-xs text-gray-500 dark:text-gray-400">
{{ t("admin.settings.ollamaCloudUsage.intervalHint") }}
</p>
</div>
</div>
<div class="flex justify-end border-t border-gray-100 pt-4 dark:border-dark-700">
<button
@@ -7934,6 +7954,7 @@ const ollamaCloudUsageSaving = ref(false);
const ollamaCloudUsageForm = reactive({
enabled: false,
interval_minutes: 60,
debounce_minutes: 1,
});
// Overload Cooldown (529) 状态
@@ -50,6 +50,7 @@ const {
getOllamaCloudUsageSettings: vi.fn().mockResolvedValue({
enabled: false,
interval_minutes: 60,
debounce_minutes: 1,
}),
updateOllamaCloudUsageSettings: vi.fn().mockImplementation(async (payload) => payload),
getGroups: vi.fn(),
@@ -647,6 +648,7 @@ describe("admin SettingsView payment visible method controls", () => {
getOllamaCloudUsageSettings.mockResolvedValue({
enabled: false,
interval_minutes: 60,
debounce_minutes: 1,
});
updateOllamaCloudUsageSettings.mockImplementation(async (payload) => payload);
getGroups.mockResolvedValue([]);
@@ -1007,6 +1009,7 @@ describe("admin SettingsView payment visible method controls", () => {
expect(card.find('[data-testid="ollama-cloud-usage-global-interval"]').exists()).toBe(false);
await card.get('[data-testid="ollama-cloud-usage-global-enabled"]').setValue(true);
await card.get('[data-testid="ollama-cloud-usage-global-debounce"]').setValue(3);
await card.get('[data-testid="ollama-cloud-usage-global-interval"]').setValue(90);
await card.get('[data-testid="ollama-cloud-usage-global-save"]').trigger("click");
await flushPromises();
@@ -1014,6 +1017,7 @@ describe("admin SettingsView payment visible method controls", () => {
expect(updateOllamaCloudUsageSettings).toHaveBeenCalledWith({
enabled: true,
interval_minutes: 90,
debounce_minutes: 3,
});
});