fix(ollama): 修复 PG<=16 上 due 判定失效并恢复抓取下限

#4850 的用量刷新调度有三个问题,本次一并修复。

jsonpath .datetime() 直到 PostgreSQL 17 才接受 ISO-8601 的 Z 标识符,
而服务写入的快照时间戳全部是 UTC(即 Z 形式)。在 PG 14/15/16 上
ollamaCloudUsageParseRFC3339SQL 因此把 fetched_at / last_attempt_at /
next_refresh_at 全部解析成 NULL,due 判定整条退化为 fail-open 分支:
每轮 20 个刷新额度被 id 最小的非 due 组占满,id 更大的组永远不会刷新——
正是该 PR 声称修复的饥饿场景。项目文档声明支持 PG 14/15/16,而集成测试
harness 固定使用 postgres:18.1,因此现有测试无法发现。

修法是在送入 jsonpath 前把结尾的 Z 改写为 +00:00。仍保留 jsonpath 而不
直接 ::timestamptz,因为通过形状正则但日历非法的值(如 2026-02-30)需要
fail-open 成 NULL 而非中断整条查询。已在真实 PG 14/15/16/17/18 上验证
五个版本行为一致,且非法日历与垃圾输入仍正确 fail-open。

成功路径不再查阅 next_refresh_at,而 nextOllamaCloudUsageDelay 的
15 分钟下限正作用于该字段,导致同组对 ollama.com 的抓取下限从 15 分钟
降到一个 runner 周期。请求间隔略大于 debounce 的交互式流量(典型 Claude
Code 用法)会把单组 24 小时抓取次数从 24-96 抬高到数百次。改为对成功路径
显式施加 fetched_at + OllamaCloudUsageMinFetchInterval 的下限,Go 与 SQL
两侧同步;空闲账号不再轮询这一主要收益不受影响。

debounce_minutes 与 interval_minutes 此前各自独立校验,因此
debounce >= interval 是合法组合;此时 min(lastUsed+debounce,
fetchedAt+maxWait) 中的 debounce 项恒为死项,管理员配置被静默忽略。
改为在写入时拒绝该组合。

另修复 refreshAccount 中 ListOllamaCloudUsageGroupAccounts 的错误被
静默吞掉(违反 CLAUDE.md 禁止忽略错误):失败时回退到更窄的活动信号会
改变 due 语义,现在记录日志。

测试:
- 恢复被删除的 7/8/9 位小数秒解析覆盖,并改写为可判别形式(断言"不应
  返回",解析失败会落入 fail-open 而被捕获)。已验证该测试在 PG 15 上
  无此修复时失败、有修复时通过。
- 新增 min fetch interval 下限与 debounce/interval 交叉校验的单元测试。
- integration harness 新增 SUB2API_TEST_POSTGRES_IMAGE 覆盖,使套件可
  针对最低受支持版本运行。
This commit is contained in:
shaw
2026-07-25 18:34:37 +08:00
parent bb0c38306f
commit 1763db3a2d
6 changed files with 212 additions and 12 deletions
@@ -356,15 +356,32 @@ func canonicalJSON(raw string) string {
// ollamaCloudUsageParseRFC3339SQL reuses the verified RFC3339(/Nano) parse path
// for a snapshot timestamp expression. Invalid or missing values fail open to NULL.
//
// The value is rewritten twice before it reaches jsonpath:
// 1. Sub-second precision beyond 6 digits is truncated, because .datetime()
// rejects more than microsecond resolution while Go emits 9 digits.
// 2. A trailing "Z" is rewritten to "+00:00". jsonpath .datetime() only learned
// to accept the ISO-8601 "Z" designator in PostgreSQL 17, and every timestamp
// this service writes is UTC (hence "Z"). Without this rewrite the parse
// silently yields NULL on PostgreSQL <= 16, which makes every due column NULL
// and collapses ListDueOllamaCloudUsageAccounts into its fail-open branch.
//
// jsonpath (rather than a direct ::timestamptz cast) is required so that values
// passing the shape regex but naming an impossible date (e.g. 2026-02-30) fail
// open to NULL instead of aborting the whole query.
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'
regexp_replace(
` + expression + `,
'(\.[0-9]{6})[0-9]+(Z|[+-][0-9]{2}:[0-9]{2})$',
'\1\2'
),
'Z$',
'+00:00'
)),
'$.datetime()', '{}'::jsonb, true
) #>> '{}'
@@ -378,9 +395,10 @@ func ollamaCloudUsageParseRFC3339SQL(expression string) string {
// 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:
// Rules mirror service.ollamaCloudUsageAutoRefreshDueAt (keep both in sync):
// - missing/invalid snapshot or times → fail-open first due
// - success: activity after fetched_at; due_at = LEAST(last_used+debounce, fetched+maxWait)
// - success: activity after fetched_at;
// due_at = GREATEST(LEAST(last_used+debounce, fetched+maxWait), fetched+minFetchInterval)
// - 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(
@@ -403,6 +421,7 @@ func (r *accountRepository) ListDueOllamaCloudUsageAccounts(
}
debounceSeconds := debounce.Seconds()
maxWaitSeconds := maxWait.Seconds()
minFetchIntervalSeconds := service.OllamaCloudUsageMinFetchInterval.Seconds()
rows, err := r.sql.QueryContext(ctx, `
WITH eligible AS (
SELECT id,
@@ -444,9 +463,12 @@ func (r *accountRepository) ListDueOllamaCloudUsageAccounts(
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)
THEN GREATEST(
LEAST(
group_last_used_at + make_interval(secs => $2::double precision),
parsed_fetched_at::timestamptz + make_interval(secs => $3::double precision)
),
parsed_fetched_at::timestamptz + make_interval(secs => $5::double precision)
)
WHEN status IN ('failed', 'unauthorized')
AND parsed_last_attempt_at IS NOT NULL
@@ -490,7 +512,7 @@ func (r *accountRepository) ListDueOllamaCloudUsageAccounts(
WHERE group_rank = 1
ORDER BY due_class, due_at NULLS FIRST, id
LIMIT $4
`, now.UTC(), debounceSeconds, maxWaitSeconds, limit)
`, now.UTC(), debounceSeconds, maxWaitSeconds, limit, minFetchIntervalSeconds)
if err != nil {
return nil, err
}
@@ -72,6 +72,81 @@ func TestListDueOllamaCloudUsageAccountsOrderingLimitAndProxyHydration(t *testin
require.Equal(t, proxy.URL(), accounts[0].Proxy.URL())
}
// TestListDueOllamaCloudUsageAccountsParsesAllRFC3339Precisions pins the SQL
// timestamp parse path across the sub-second precisions and zone spellings that
// actually reach the database.
//
// Each fixture stores a fetched_at only two minutes old with activity 30s later,
// so a correctly parsed row is NOT due (debounce and the min fetch interval both
// place it in the future). A row whose timestamp fails to parse becomes NULL and
// falls into the fail-open branch, which makes it due. Asserting on absence is
// therefore what makes this test able to fail:
//
// - Go writes UTC times, i.e. the "Z" designator. jsonpath .datetime() only
// accepts "Z" from PostgreSQL 17 on, so without the Z -> +00:00 rewrite in
// ollamaCloudUsageParseRFC3339SQL every fixture here goes due on 14-16.
// - 7/8/9 sub-second digits exceed the microsecond resolution .datetime()
// allows and must be truncated first.
//
// Run against the oldest supported server to exercise the version-sensitive path:
//
// SUB2API_TEST_POSTGRES_IMAGE=postgres:15-alpine go test -tags integration ./internal/repository/
func TestListDueOllamaCloudUsageAccountsParsesAllRFC3339Precisions(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)
activity := now.Add(-30 * time.Second)
// All three spell the same instant, now-2m, with different precision/zone.
notDue := map[string]string{
"nano-z": "2026-07-22T13:58:00.123456789Z",
"eight-positive": "2026-07-22T14:58:00.12345678+01:00",
"seven-negative": "2026-07-22T11:58:00.1234567-02:00",
}
for name, fetchedAt := range notDue {
_ = mustCreateAccount(t, tx.Client(), &service.Account{
Name: "ollama-precision-" + name, Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "precision-" + 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": fetchedAt,
"last_attempt_at": fetchedAt,
},
},
LastUsedAt: &activity,
})
}
// Guards against a vacuous pass: an genuinely due row must still come back.
staleFetched := now.Add(-2 * time.Hour)
due := mustCreateAccount(t, tx.Client(), &service.Account{
Name: "ollama-precision-due", Platform: service.PlatformOpenAI, Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "precision-due", "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": staleFetched.UTC().Format(time.RFC3339Nano),
"last_attempt_at": staleFetched.UTC().Format(time.RFC3339Nano),
},
},
LastUsedAt: &activity,
})
accounts, err := repo.ListDueOllamaCloudUsageAccounts(ctx, now, time.Minute, time.Hour, 10)
require.NoError(t, err)
ids := accountIDs(accounts)
require.Contains(t, ids, due.ID, "a stale snapshot with fresh activity must be due")
require.Len(t, ids, 1,
"only the stale group may be due; extra rows mean a timestamp failed to parse and fell into the fail-open branch")
}
func TestListDueOllamaCloudUsageAccountsUsesGroupMaxLastUsedAndFailsOpen(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
@@ -194,7 +194,7 @@ func TestListDueOllamaCloudUsageAccountsFiltersOrdersAndLimits(t *testing.T) {
maxWait := time.Hour
var capturedSQL string
mock.ExpectQuery("WITH eligible AS").
WithArgs(now.UTC(), debounce.Seconds(), maxWait.Seconds(), 20).
WithArgs(now.UTC(), debounce.Seconds(), maxWait.Seconds(), 20, service.OllamaCloudUsageMinFetchInterval.Seconds()).
WillReturnRows(sqlmock.NewRows([]string{"id", "group_last_used_at"}))
repo := newAccountRepositoryWithSQL(nil, captureQuerySQL{db: db, captured: &capturedSQL}, nil)
@@ -217,6 +217,13 @@ func TestListDueOllamaCloudUsageAccountsFiltersOrdersAndLimits(t *testing.T) {
"LIMIT $4",
"make_interval(secs => $2::double precision)",
"make_interval(secs => $3::double precision)",
// Minimum interval floor between successful fetches.
"make_interval(secs => $5::double precision)",
// jsonpath .datetime() only accepts the ISO-8601 "Z" designator from
// PostgreSQL 17 on, and this service writes UTC timestamps. Without this
// rewrite every parsed_* column is NULL on 14-16 and the due filter
// collapses into its fail-open branch.
`regexp_replace( regexp_replace( fetched_at, '(\.[0-9]{6})[0-9]+(Z|[+-][0-9]{2}:[0-9]{2})$', '\1\2' ), 'Z$', '+00:00' )`,
"group_last_used_at > parsed_fetched_at::timestamptz",
"group_last_used_at > parsed_last_attempt_at::timestamptz",
"$1 >= activity_due_at",
@@ -140,7 +140,20 @@ func dockerIsAvailable(ctx context.Context) bool {
return cmd.Run() == nil
}
// selectDockerImage resolves the container image for the harness.
//
// SUB2API_TEST_POSTGRES_IMAGE overrides the PostgreSQL image so the suite can be
// run against the oldest documented-supported server, not just the newest. That
// matters for SQL that behaves differently across major versions: jsonpath
// .datetime() only accepts the ISO-8601 "Z" designator from PostgreSQL 17 on, so
// a suite pinned to 18 cannot observe breakage on 14-16.
//
// SUB2API_TEST_POSTGRES_IMAGE=postgres:15-alpine go test -tags integration ./internal/repository/
func selectDockerImage(ctx context.Context, preferred string) string {
if override := strings.TrimSpace(os.Getenv("SUB2API_TEST_POSTGRES_IMAGE")); override != "" &&
strings.HasPrefix(preferred, "postgres:") {
return override
}
if dockerImageExists(ctx, preferred) {
return preferred
}
+35 -2
View File
@@ -32,6 +32,13 @@ const (
OllamaCloudUsageAutoRefreshExtraKey = "ollama_cloud_usage_auto_refresh"
OllamaCloudUsageSnapshotExtraKey = "ollama_cloud_usage_snapshot"
// OllamaCloudUsageMinFetchInterval is the hard floor between two successful
// fetches of the same group, mirroring the floor nextOllamaCloudUsageDelay
// applies to next_refresh_at. Activity may bring a refresh forward to this
// bound but never past it. Exported so the repository can apply the same
// floor inside the SQL due filter.
OllamaCloudUsageMinFetchInterval = ollamaCloudUsageMinIntervalMinutes * time.Minute
ollamaCloudUsageSettingsURL = "https://ollama.com/settings"
ollamaCloudUsageDefaultIntervalMinutes = 60
ollamaCloudUsageMinIntervalMinutes = 15
@@ -212,6 +219,15 @@ func (s *SettingService) SetOllamaCloudUsageSettings(ctx context.Context, settin
fmt.Sprintf("debounce_minutes must be between %d and %d", ollamaCloudUsageMinDebounceMinutes, ollamaCloudUsageMaxDebounceMinutes),
)
}
// The due time is min(lastUsed+debounce, fetchedAt+maxWait). Once the debounce
// reaches the max wait the debounce term can never win, so the knob would be
// silently inert instead of doing what the operator asked for.
if settings.DebounceMinutes >= settings.IntervalMinutes {
return infraerrors.BadRequest(
"INVALID_OLLAMA_CLOUD_USAGE_DEBOUNCE",
fmt.Sprintf("debounce_minutes (%d) must be less than interval_minutes (%d)", settings.DebounceMinutes, settings.IntervalMinutes),
)
}
normalizeOllamaCloudUsageSettings(settings)
data, err := json.Marshal(settings)
if err != nil {
@@ -301,7 +317,16 @@ func ollamaCloudUsageAutoRefreshDueAt(
return time.Time{}, false
}
lastUsed := groupLastUsedAt.UTC()
return minTime(lastUsed.Add(debounce), fetchedAt.Add(maxWait)), true
dueAt := minTime(lastUsed.Add(debounce), fetchedAt.Add(maxWait))
// Keep the pre-existing hard floor between successful fetches. The success
// path no longer consults next_refresh_at, which is where
// nextOllamaCloudUsageDelay used to apply ollamaCloudUsageMinIntervalMinutes;
// without this, request traffic spaced slightly wider than the debounce
// drives the group's outbound rate far above the previous minimum.
if floor := fetchedAt.Add(OllamaCloudUsageMinFetchInterval); dueAt.Before(floor) {
return floor, true
}
return dueAt, true
case OllamaCloudUsageStatusFailed, OllamaCloudUsageStatusUnauthorized:
if snapshot.LastAttemptAt.IsZero() {
return time.Time{}, true
@@ -801,7 +826,15 @@ func (s *OllamaCloudUsageService) refreshAccount(ctx context.Context, accountID
}
groupLastUsed := account.LastUsedAt
if writer, ok := s.accountRepo.(ollamaCloudUsageRepository); ok {
if siblings, listErr := writer.ListOllamaCloudUsageGroupAccounts(ctx, []*Account{account}); listErr == nil {
siblings, listErr := writer.ListOllamaCloudUsageGroupAccounts(ctx, []*Account{account})
if listErr != nil {
// Fall back to this account's own last_used_at. That is a narrower
// activity signal than the group maximum, so the due check may skip a
// refresh it would otherwise have run; surface it rather than
// silently changing the due semantics.
logger.LegacyPrintf("service.ollama_cloud_usage",
"group_last_used_lookup_failed: account_id=%d err=%v", account.ID, listErr)
} else {
groupLastUsed = maxOllamaCloudUsageGroupLastUsed(siblings)
}
}
@@ -333,6 +333,16 @@ func TestOllamaCloudUsageSettingsDefaultOffAndValidation(t *testing.T) {
require.Equal(t, 90, settings.IntervalMinutes)
require.Equal(t, 2, settings.DebounceMinutes)
// debounce >= interval would make the debounce term unreachable in
// min(lastUsed+debounce, fetchedAt+maxWait), silently ignoring the operator's
// setting, so it is rejected rather than accepted and dropped.
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 15, DebounceMinutes: 15})
require.Error(t, err, "debounce equal to interval must be rejected")
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 15, DebounceMinutes: 60})
require.Error(t, err, "debounce greater than interval must be rejected")
err = settingsService.SetOllamaCloudUsageSettings(context.Background(), &OllamaCloudUsageSettings{Enabled: true, IntervalMinutes: 16, DebounceMinutes: 15})
require.NoError(t, err, "debounce below interval stays valid")
// Legacy JSON without debounce_minutes defaults to 1.
repo.values[SettingKeyOllamaCloudUsageSettings] = `{"enabled":true,"interval_minutes":45}`
settings, err = settingsService.GetOllamaCloudUsageSettings(context.Background())
@@ -384,6 +394,46 @@ func TestOllamaCloudUsageIsAutoRefreshDue(t *testing.T) {
}, nil, now, debounce, maxWait), "ok without fetched_at fails open")
}
// The success path stopped consulting next_refresh_at, which is where
// nextOllamaCloudUsageDelay used to apply the minimum interval. Activity may pull
// a refresh forward only as far as that floor, otherwise request traffic spaced
// just wider than the debounce drives the group's outbound rate far above the
// pre-existing minimum.
func TestOllamaCloudUsageAutoRefreshDueAtHonoursMinFetchInterval(t *testing.T) {
debounce := time.Minute
maxWait := time.Hour
now := time.Date(2026, time.July, 25, 12, 0, 0, 0, time.UTC)
ptr := func(ts time.Time) *time.Time { return &ts }
// Debounce elapsed, but the last successful fetch is inside the floor.
recent := now.Add(-5 * time.Minute)
recentSnap := &OllamaCloudUsageSnapshot{
Status: OllamaCloudUsageStatusOK, FetchedAt: ptr(recent), LastAttemptAt: recent,
}
dueAt, ok := ollamaCloudUsageAutoRefreshDueAt(recentSnap, ptr(now.Add(-2*time.Minute)), debounce, maxWait)
require.True(t, ok)
require.Equal(t, recent.Add(OllamaCloudUsageMinFetchInterval), dueAt,
"due time must be clamped to fetched_at + min fetch interval")
require.False(t, ollamaCloudUsageIsAutoRefreshDue(recentSnap, ptr(now.Add(-2*time.Minute)), now, debounce, maxWait),
"debounce alone must not refresh within the min fetch interval")
// Once the floor has passed the debounce governs again.
atFloor := now.Add(-OllamaCloudUsageMinFetchInterval)
floorSnap := &OllamaCloudUsageSnapshot{
Status: OllamaCloudUsageStatusOK, FetchedAt: ptr(atFloor), LastAttemptAt: atFloor,
}
require.True(t, ollamaCloudUsageIsAutoRefreshDue(floorSnap, ptr(now.Add(-2*time.Minute)), now, debounce, maxWait),
"past the floor a quiet debounce window is due")
// The floor never delays a refresh that max-wait has already forced.
stale := now.Add(-2 * time.Hour)
staleSnap := &OllamaCloudUsageSnapshot{
Status: OllamaCloudUsageStatusOK, FetchedAt: ptr(stale), LastAttemptAt: stale,
}
require.True(t, ollamaCloudUsageIsAutoRefreshDue(staleSnap, ptr(now), now, debounce, maxWait),
"max-wait still forces due on a stale snapshot")
}
func TestScheduleOllamaCloudUsageActivityOnlyForOllama(t *testing.T) {
deferred := NewDeferredService(nil, nil, time.Second)
ollama := ollamaUsageAccount(1)