Respect provider rate limits and preserve bank connections on throttling

This commit is contained in:
Lars Nolden
2026-09-10 17:25:09 +02:00
parent 2259db3e85
commit ba3ea6ae5a
14 changed files with 1250 additions and 95 deletions
+30
View File
@@ -150,6 +150,36 @@ failed enrichment stays unclassified with an error visible in Transactions.
Classification requests use one transaction at a time, not batches. Known Classification requests use one transaction at a time, not batches. Known
merchant defaults can classify without any configured AI key. merchant defaults can classify without any configured AI key.
Provider rate limits
--------------------
Classification and Enable Banking use separate in-memory request gates. AI
imports and previews, including previews choosing another model, share the AI
gate. HTTP 429 retries retain the original request and privacy/authentication
controls. Other HTTP failures are not automatically retried.
Each operation makes at most four attempts. Without a usable Retry-After hint,
retry delays are 1, 2 and 4 seconds; an exhausted fourth attempt retains an
8-second cooldown. Retry-After delta-seconds and HTTP dates can extend these
delays, never shorten them. Automatic waiting totals at most two minutes,
separate from each network attempt's timeout (bank: at most 30 seconds; AI:
45 seconds by default). Waiting and queued requests honor context cancellation.
Long hints, exhausted attempts, and canceled retries retain the provider's
cooldown. New calls fail without contacting that provider before its retry
time. Local merchant rules still work. Unrepresentably large positive hints
disable automatic retries rather than overflow into an early request.
These gates live in the running client, not the persisted financial journal.
Retries cannot lift provider quotas; wait until the reported time. Existing
failed classifications require another Analyse preview and explicit Apply.
Bank GET requests may retry 429. Authorization and once-only session exchange
POST requests never replay automatically, but still establish a cooldown.
Session status checks use the authorized UID list and expiry, without repeated
account-details requests; full account metadata remains saved from Exchange.
An unavailable session reports its original safe rate error once, without
additional unavailable-account errors or a false reconnection requirement.
Failed-account cursors and the last successful complete sync remain unchanged.
Enable Banking Enable Banking
-------------- --------------
Register your application and public certificate with Enable Banking. For Register your application and public certificate with Enable Banking. For
+4
View File
@@ -142,6 +142,8 @@ Finance Duck verifies the callback state, exchanges the returned code for a `ses
Initial synchronization requests the selected number of **calendar months of booked transactions per account**, defaulting to **12 months**. The bank may provide less history. The choice is saved with the bank connection and reused on reconnection. Subsequent daily synchronization overlaps each account's last successful sync by **14 days**. **Sync now** starts a manual synchronization. Existing accounts keep their successful-sync cursors: changing the history choice or reconnecting does **not** backfill them. Older history can be imported with CSV. Initial synchronization requests the selected number of **calendar months of booked transactions per account**, defaulting to **12 months**. The bank may provide less history. The choice is saved with the bank connection and reused on reconnection. Subsequent daily synchronization overlaps each account's last successful sync by **14 days**. **Sync now** starts a manual synchronization. Existing accounts keep their successful-sync cursors: changing the history choice or reconnecting does **not** backfill them. Older history can be imported with CSV.
**HTTP 429 is a provider rate limit, not evidence that bank consent has expired.** Bank reads honor `Retry-After` and use bounded exponential retries. A longer or exhausted limit pauses further requests until the reported retry time; failed accounts keep their previous sync cursors and imported data. Session checks use the saved account metadata rather than fetching every account's details again. A failed session is reported once instead of also marking each of its accounts unavailable. After the cooldown, **Sync now** can retry; the warning clears after a successful sync. One-time authorization and code-exchange requests are never automatically replayed.
### Import older history for a connected account ### Import older history for a connected account
Open **Accounts**, find the account, and click **Import older history**. Choose **Months back** (default **12**, whole numbers from **1 to 120**) and confirm. This requests that account's booked transactions from the selected number of calendar months ago through today; the bank may provide less history. Open **Accounts**, find the account, and click **Import older history**. Choose **Months back** (default **12**, whole numbers from **1 to 120**) and confirm. This requests that account's booked transactions from the selected number of calendar months ago through today; the bank may provide less history.
@@ -359,6 +361,8 @@ Bank imports do **not** require this key. Without AI, explicit merchant-default
Every AI classification requests `provider.data_collection = "deny"`, `provider.zdr = true`, and `provider.require_parameters = true`. Unsupported private routing fails rather than falling back to a less restrictive provider. Amount sharing is off by default. Keep OpenRouter account prompt logging disabled as well. Automatic redaction minimizes data; it is not a guarantee that arbitrary transaction prose is anonymous. Every AI classification requests `provider.data_collection = "deny"`, `provider.zdr = true`, and `provider.require_parameters = true`. Unsupported private routing fails rather than falling back to a less restrictive provider. Amount sharing is off by default. Keep OpenRouter account prompt logging disabled as well. Automatic redaction minimizes data; it is not a guarantee that arbitrary transaction prose is anonymous.
Classification honors HTTP 429 `Retry-After` delays (seconds or HTTP dates), with exponential backoff when no usable delay is supplied. It makes at most **four attempts**, with at most **two minutes of automatic waiting**, preserving the same model, sanitized prompt, and privacy controls on every attempt. Imports and classification previews share a cooldown, so remaining records cannot immediately hammer a rate-limited provider. Long or exhausted limits leave records unclassified with a retry-time error; local merchant rules still work. After the cooldown, run **AI classification → Analyse** again for previously failed records—repeating a bank import does not reclassify existing transactions.
## Data, backups, and recovery ## Data, backups, and recovery
Back up the **entire canonical finance directory**, including registry files, journals, `config.toml` when present, and operational/recovery state, plus any separately stored environment-managed secrets. `state/openrouter.json` and `state/enablebanking.json` contain UI-managed credentials: protect backups accordingly, including the matching banking session state. Stop the service for a consistent filesystem backup. DuckDB under `cache/` can be excluded and rebuilt. Back up the **entire canonical finance directory**, including registry files, journals, `config.toml` when present, and operational/recovery state, plus any separately stored environment-managed secrets. `state/openrouter.json` and `state/enablebanking.json` contain UI-managed credentials: protect backups accordingly, including the matching banking session state. Stop the service for a consistent filesystem backup. DuckDB under `cache/` can be excluded and rebuilt.
+87
View File
@@ -4,11 +4,14 @@ import (
"context" "context"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"reflect" "reflect"
"strings" "strings"
"sync/atomic"
"testing" "testing"
"time"
"finance-duck/internal/analytics" "finance-duck/internal/analytics"
"finance-duck/internal/classification" "finance-duck/internal/classification"
@@ -86,6 +89,90 @@ func TestFailedClassificationStillImportsAndRetryIsIdempotent(t *testing.T) {
t.Fatalf("import not visible in analytics: %+v", dash.Totals) t.Fatalf("import not visible in analytics: %+v", dash.Totals)
} }
} }
func TestPreviewCooldownProtectsLaterPreviewsAndImports(t *testing.T) {
a, s := testApp(t)
s = seed(t, a, s)
before := domain.Clone(s.Data)
var calls atomic.Int32
provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
calls.Add(1)
w.Header().Set("Retry-After", "300")
w.WriteHeader(http.StatusTooManyRequests)
}))
defer provider.Close()
a.classifier = classification.Client{APIKey: "test", Model: "test/model", BaseURL: provider.URL}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
for _, model := range []string{"test/model", "test/another-model"} {
preview, err := a.Preview(ctx, PreviewRequest{
Revision: s.Revision, From: "2026-09-01", To: "2026-09-30",
Model: model, Fields: Fields{Category: true},
})
if err != nil {
t.Fatal(err)
}
if preview.Analysed != 2 || len(preview.Errors) != 2 || len(preview.Changes) != 0 {
t.Fatalf("rate-limited preview did not preserve both records: %+v", preview)
}
}
unchanged, err := a.Snapshot(ctx)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(before, unchanged.Data) {
t.Fatal("rate-limited previews changed canonical data")
}
a.mu.Lock()
result, err := a.importFacts(ctx, unchanged, []domain.Facts{sampleFacts("ALDI", "2026-09-10", "-12.34")})
a.mu.Unlock()
if err != nil {
t.Fatal(err)
}
if result.Imported != 1 || len(result.State.Data.Transactions) != 3 {
t.Fatal("provider cooldown lost the newly imported record")
}
for _, tx := range result.State.Data.Transactions {
if tx.Facts.ExternalID == hex.EncodeToString([]byte("ALDI")) {
if tx.Enrichment.Classification.Error == "" || tx.Enrichment.CategoryID != domain.ExpenseFallback {
t.Fatal("cooldown did not leave imported facts editable and unclassified")
}
}
}
if got := calls.Load(); got != 1 {
t.Fatalf("previews and imports bypassed shared provider cooldown: %d requests", got)
}
}
func TestCancelledLastClassificationDoesNotProducePreview(t *testing.T) {
a, s := testApp(t)
s = seed(t, a, s)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Retry-After", "60")
w.WriteHeader(http.StatusTooManyRequests)
cancel()
}))
defer provider.Close()
a.classifier = classification.Client{APIKey: "test", Model: "test/model", BaseURL: provider.URL}
p, err := a.Preview(ctx, PreviewRequest{
Revision: s.Revision, From: "2026-09-09", To: "2026-09-09",
Model: "test/model", Fields: Fields{Category: true},
})
if !errors.Is(err, context.Canceled) || p.ID != "" {
t.Fatalf("cancelled final record produced a preview: id=%q, error=%v", p.ID, err)
}
after, err := a.Snapshot(context.Background())
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(s.Data, after.Data) {
t.Fatal("cancelled preview changed canonical data")
}
}
func mockClassifier(t *testing.T, a *App, inspect ...func(*http.Request)) { func mockClassifier(t *testing.T, a *App, inspect ...func(*http.Request)) {
t.Helper() t.Helper()
mock := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { mock := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+11 -4
View File
@@ -22,10 +22,10 @@ type backfillBank struct {
fetchErr error fetchErr error
} }
func (b *backfillBank) Status(ctx context.Context, id string) (banking.Session, error) { func (b *backfillBank) Status(ctx context.Context, id string) (banking.SessionStatus, error) {
b.statusIDs = append(b.statusIDs, id) b.statusIDs = append(b.statusIDs, id)
if b.statusErr != nil { if b.statusErr != nil {
return banking.Session{}, b.statusErr return banking.SessionStatus{}, b.statusErr
} }
return b.historyBank.Status(ctx, id) return b.historyBank.Status(ctx, id)
} }
@@ -204,7 +204,7 @@ func TestBackfillRejectsUnsafePrerequisitesBeforeProviderContact(t *testing.T) {
} }
func TestBackfillProviderFailuresDoNotImportPartialData(t *testing.T) { func TestBackfillProviderFailuresDoNotImportPartialData(t *testing.T) {
for _, scenario := range []string{"not configured", "unavailable", "revoked", "provider expired", "account absent", "partial retrieval"} { for _, scenario := range []string{"not configured", "unavailable", "revoked", "provider expired", "account absent", "partial retrieval", "status rate limit", "retrieval rate limit"} {
t.Run(scenario, func(t *testing.T) { t.Run(scenario, func(t *testing.T) {
a, s, b := backfillApp(t) a, s, b := backfillApp(t)
s = seed(t, a, s) s = seed(t, a, s)
@@ -221,6 +221,10 @@ func TestBackfillProviderFailuresDoNotImportPartialData(t *testing.T) {
b.session.Accounts = []domain.Account{{ID: s.Data.Accounts[0].ID, ExternalAccountID: "wrong_uid"}} b.session.Accounts = []domain.Account{{ID: s.Data.Accounts[0].ID, ExternalAccountID: "wrong_uid"}}
case "partial retrieval": case "partial retrieval":
b.fetchErr = errors.New("private provider response") b.fetchErr = errors.New("private provider response")
case "status rate limit":
b.statusErr = bankRateError(t)
case "retrieval rate limit":
b.fetchErr = bankRateError(t)
} }
beforeOps, err := json.Marshal(a.ops) beforeOps, err := json.Marshal(a.ops)
if err != nil { if err != nil {
@@ -233,7 +237,10 @@ func TestBackfillProviderFailuresDoNotImportPartialData(t *testing.T) {
if strings.Contains(err.Error(), "private provider response") { if strings.Contains(err.Error(), "private provider response") {
t.Fatal("provider error exposed private response data") t.Fatal("provider error exposed private response data")
} }
if scenario != "partial retrieval" && len(b.accounts) != 0 { if strings.Contains(scenario, "rate limit") && (!strings.Contains(err.Error(), "429") || !strings.Contains(err.Error(), "retry") || errors.Is(err, banking.ErrReconnect)) {
t.Fatal("backfill obscured rate limiting or falsely required reconnect")
}
if scenario != "partial retrieval" && scenario != "retrieval rate limit" && len(b.accounts) != 0 {
t.Fatal("unavailable consent reached transaction retrieval") t.Fatal("unavailable consent reached transaction retrieval")
} }
current, err := a.Snapshot(context.Background()) current, err := a.Snapshot(context.Background())
+3 -3
View File
@@ -185,11 +185,11 @@ func TestRenewedConsentWakesSchedulerAndAutomaticallyImports(t *testing.T) {
type recoveryBank struct{ historyBank } type recoveryBank struct{ historyBank }
func (b *recoveryBank) Status(ctx context.Context, id string) (banking.Session, error) { func (b *recoveryBank) Status(ctx context.Context, id string) (banking.SessionStatus, error) {
if id != b.session.ID { if id != b.session.ID {
return banking.Session{}, banking.ErrReconnect return banking.SessionStatus{}, banking.ErrReconnect
} }
return b.session, nil return b.historyBank.Status(ctx, id)
} }
func TestInterruptedRenewalDiscardsSupersededConsentDuringRecovery(t *testing.T) { func TestInterruptedRenewalDiscardsSupersededConsentDuringRecovery(t *testing.T) {
+47 -19
View File
@@ -12,6 +12,7 @@ import (
"finance-duck/internal/banking" "finance-duck/internal/banking"
"finance-duck/internal/classification" "finance-duck/internal/classification"
"finance-duck/internal/domain" "finance-duck/internal/domain"
"finance-duck/internal/ratelimit"
) )
type ImportResult struct { type ImportResult struct {
@@ -136,27 +137,19 @@ func (a *App) Backfill(ctx context.Context, rev, accountID string, historyMonths
} }
current, err := a.bank.Status(ctx, session.ID) current, err := a.bank.Status(ctx, session.ID)
if err != nil { if err != nil {
if errors.Is(err, banking.ErrReconnect) { return ImportResult{}, bankFailure(err, "bank connection unavailable; retry importing history")
return ImportResult{}, banking.ErrReconnect
}
return ImportResult{}, errors.New("bank connection unavailable; retry importing history")
} }
expiry, err = time.Parse(time.RFC3339, current.ValidUntil) expiry, err = time.Parse(time.RFC3339, current.ValidUntil)
if err != nil || !expiry.After(time.Now()) { if err != nil || !expiry.After(time.Now()) {
return ImportResult{}, banking.ErrReconnect return ImportResult{}, banking.ErrReconnect
} }
if !slices.ContainsFunc(current.Accounts, func(linked domain.Account) bool { if !slices.Contains(current.AccountIDs, account.ExternalAccountID) {
return linked.ExternalAccountID == account.ExternalAccountID
}) {
return ImportResult{}, banking.ErrReconnect return ImportResult{}, banking.ErrReconnect
} }
now := time.Now().UTC() now := time.Now().UTC()
facts, err := a.bank.Transactions(ctx, account, now.AddDate(0, -historyMonths, 0).Format("2006-01-02"), now.Format("2006-01-02")) facts, err := a.bank.Transactions(ctx, account, now.AddDate(0, -historyMonths, 0).Format("2006-01-02"), now.Format("2006-01-02"))
if err != nil { if err != nil {
if errors.Is(err, banking.ErrReconnect) { return ImportResult{}, bankFailure(err, "transaction retrieval failed; retry importing history")
return ImportResult{}, banking.ErrReconnect
}
return ImportResult{}, errors.New("transaction retrieval failed; retry importing history")
} }
// Use normal import processing without changing sync cursors or saved consent // Use normal import processing without changing sync cursors or saved consent
// settings, including when the requested range adds no transactions. // settings, including when the requested range adds no transactions.
@@ -302,6 +295,20 @@ func (a *App) Balances(ctx context.Context, id string) ([]banking.Balance, error
} }
return nil, errors.New("account is not connected") return nil, errors.New("account is not connected")
} }
// Only typed, locally generated errors are safe to expose; provider errors may
// wrap private response data even when their underlying cause is recognizable.
func bankFailure(err error, fallback string) error {
var limited *ratelimit.RateLimitError
if errors.As(err, &limited) {
return fmt.Errorf("Enable Banking: %w", limited)
}
if errors.Is(err, banking.ErrReconnect) {
return banking.ErrReconnect
}
return errors.New(fallback)
}
func (a *App) Sync(ctx context.Context) (State, error) { func (a *App) Sync(ctx context.Context) (State, error) {
a.mu.Lock() a.mu.Lock()
defer a.mu.Unlock() defer a.mu.Unlock()
@@ -343,25 +350,35 @@ func (a *App) Sync(ctx context.Context) (State, error) {
} }
validAccounts := map[string]bool{} validAccounts := map[string]bool{}
accountSession := map[string]string{} accountSession := map[string]string{}
failedSessions := map[string]bool{}
for i, session := range a.ops.Sessions { for i, session := range a.ops.Sessions {
for _, account := range session.Accounts { for _, account := range session.Accounts {
accountSession[account.ID] = session.ID accountSession[account.ID] = session.ID
} }
meta := a.ops.Consents[session.ID] meta := a.ops.Consents[session.ID]
current, e := a.bank.Status(ctx, session.ID) current, e := a.bank.Status(ctx, session.ID)
if e == nil {
expiry, parseErr := time.Parse(time.RFC3339, current.ValidUntil)
if parseErr != nil || !expiry.After(time.Now()) {
e = banking.ErrReconnect
}
}
if e != nil { if e != nil {
meta.Error = e.Error() meta.Error = bankFailure(e, "bank connection unavailable; retry synchronization").Error()
meta.NeedsReconnect = errors.Is(e, banking.ErrReconnect) meta.NeedsReconnect = errors.Is(e, banking.ErrReconnect)
a.ops.Consents[session.ID] = meta a.ops.Consents[session.ID] = meta
failures = append(failures, meta.Institution+": "+meta.Error) failures = append(failures, meta.Institution+": "+meta.Error)
failedSessions[session.ID] = true
continue continue
} }
meta.Error = "" meta.Error = ""
meta.NeedsReconnect = false meta.NeedsReconnect = false
a.ops.Consents[session.ID] = meta a.ops.Consents[session.ID] = meta
a.ops.Sessions[i].ValidUntil = current.ValidUntil a.ops.Sessions[i].ValidUntil = current.ValidUntil
for _, account := range current.Accounts { for _, account := range session.Accounts {
validAccounts[account.ExternalAccountID] = true if account.ExternalAccountID != "" && slices.Contains(current.AccountIDs, account.ExternalAccountID) {
validAccounts[account.ID] = true
}
} }
} }
now := time.Now().UTC() now := time.Now().UTC()
@@ -370,8 +387,18 @@ func (a *App) Sync(ctx context.Context) (State, error) {
if !account.Active || account.ExternalAccountID == "" { if !account.Active || account.ExternalAccountID == "" {
continue continue
} }
if !validAccounts[account.ExternalAccountID] { sessionID := accountSession[account.ID]
if failedSessions[sessionID] {
continue
}
if !validAccounts[account.ID] {
failures = append(failures, account.DisplayName+": bank connection unavailable") failures = append(failures, account.DisplayName+": bank connection unavailable")
if sessionID != "" {
meta := a.ops.Consents[sessionID]
meta.Error = banking.ErrReconnect.Error()
meta.NeedsReconnect = true
a.ops.Consents[sessionID] = meta
}
continue continue
} }
var from string var from string
@@ -383,10 +410,11 @@ func (a *App) Sync(ctx context.Context) (State, error) {
} }
facts, e := a.bank.Transactions(ctx, account, from, to) facts, e := a.bank.Transactions(ctx, account, from, to)
if e != nil { if e != nil {
meta := a.ops.Consents[accountSession[account.ID]] meta := a.ops.Consents[sessionID]
meta.Error = "Transaction retrieval failed; retry synchronization" meta.Error = bankFailure(e, "transaction retrieval failed; retry synchronization").Error()
a.ops.Consents[accountSession[account.ID]] = meta meta.NeedsReconnect = meta.NeedsReconnect || errors.Is(e, banking.ErrReconnect)
failures = append(failures, account.DisplayName+": transaction retrieval failed") a.ops.Consents[sessionID] = meta
failures = append(failures, account.DisplayName+": "+meta.Error)
continue continue
} }
result, e := a.importFacts(ctx, s, facts) result, e := a.importFacts(ctx, s, facts)
+4 -2
View File
@@ -70,7 +70,7 @@ func (a *App) Preview(ctx context.Context, r PreviewRequest) (Preview, error) {
} }
a.mu.Lock() a.mu.Lock()
s, err := a.snapshot(ctx) s, err := a.snapshot(ctx)
client := a.classifier client := a.classifier.WithModel(r.Model)
a.mu.Unlock() a.mu.Unlock()
if err != nil { if err != nil {
return Preview{}, err return Preview{}, err
@@ -78,7 +78,6 @@ func (a *App) Preview(ctx context.Context, r PreviewRequest) (Preview, error) {
if r.Revision != s.Revision { if r.Revision != s.Revision {
return Preview{}, errors.New("revision conflict: reload before analysing") return Preview{}, errors.New("revision conflict: reload before analysing")
} }
client.Model = r.Model
p := Preview{ID: domain.NewID("preview"), Revision: s.Revision, Changes: []Change{}, Errors: []ClassificationError{}, created: time.Now()} p := Preview{ID: domain.NewID("preview"), Revision: s.Revision, Changes: []Change{}, Errors: []ClassificationError{}, created: time.Now()}
baseMerchants := len(s.Data.Merchants) baseMerchants := len(s.Data.Merchants)
for _, t := range s.Data.Transactions { for _, t := range s.Data.Transactions {
@@ -90,6 +89,9 @@ func (a *App) Preview(ctx context.Context, r PreviewRequest) (Preview, error) {
} }
p.Analysed++ p.Analysed++
proposal, e := client.Classify(ctx, t.Facts, s.Data, true) proposal, e := client.Classify(ctx, t.Facts, s.Data, true)
if err = ctx.Err(); err != nil {
return Preview{}, err
}
if e != nil { if e != nil {
p.Errors = append(p.Errors, ClassificationError{t.Facts.ID, e.Error()}) p.Errors = append(p.Errors, ClassificationError{t.Facts.ID, e.Error()})
continue continue
+205 -3
View File
@@ -3,12 +3,17 @@ package app
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"io"
"net/http"
"reflect" "reflect"
"strings"
"testing" "testing"
"time" "time"
"finance-duck/internal/banking" "finance-duck/internal/banking"
"finance-duck/internal/domain" "finance-duck/internal/domain"
"finance-duck/internal/ratelimit"
) )
type bankScenario struct { type bankScenario struct {
@@ -22,11 +27,15 @@ func (b *bankScenario) Authorize(context.Context, string, string, string) (strin
func (b *bankScenario) Exchange(context.Context, string) (banking.Session, error) { func (b *bankScenario) Exchange(context.Context, string) (banking.Session, error) {
return b.session, nil return b.session, nil
} }
func (b *bankScenario) Status(context.Context, string) (banking.Session, error) { func (b *bankScenario) Status(context.Context, string) (banking.SessionStatus, error) {
if b.fail { if b.fail {
return banking.Session{}, errors.New("expired") return banking.SessionStatus{}, errors.New("expired")
} }
return b.session, nil status := banking.SessionStatus{ValidUntil: b.session.ValidUntil}
for _, account := range b.session.Accounts {
status.AccountIDs = append(status.AccountIDs, account.ExternalAccountID)
}
return status, nil
} }
func (b *bankScenario) Balances(context.Context, string) ([]banking.Balance, error) { func (b *bankScenario) Balances(context.Context, string) ([]banking.Balance, error) {
return []banking.Balance{{Amount: "100.00", Currency: "EUR", Type: "CLBD"}}, nil return []banking.Balance{{Amount: "100.00", Currency: "EUR", Type: "CLBD"}}, nil
@@ -110,3 +119,196 @@ func TestReconnectReplacesOldConsentWithoutDuplicatingLocalAccount(t *testing.T)
t.Fatal("authorization state replay was accepted") t.Fatal("authorization state replay was accepted")
} }
} }
// Exercise the controller's typed error without contacting a provider or waiting.
func bankRateError(t *testing.T) error {
t.Helper()
var controller ratelimit.Controller
ctx := context.Background()
if err := controller.Acquire(ctx); err != nil {
t.Fatal(err)
}
defer controller.Release()
_, err := controller.Do(ctx, func(context.Context) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusTooManyRequests,
Header: http.Header{"Retry-After": []string{"300"}},
Body: io.NopCloser(strings.NewReader("private provider response")),
}, nil
}, false)
if err == nil {
t.Fatal("rate limit response was accepted")
}
return fmt.Errorf("private provider response: %w", err)
}
type sessionBank struct {
bankScenario
statuses map[string]banking.SessionStatus
failures map[string]error
fetched []string
}
func (b *sessionBank) Status(_ context.Context, id string) (banking.SessionStatus, error) {
return b.statuses[id], b.failures[id]
}
func (b *sessionBank) Transactions(ctx context.Context, account domain.Account, from, to string) ([]domain.Facts, error) {
b.fetched = append(b.fetched, account.ID)
return b.bankScenario.Transactions(ctx, account, from, to)
}
func TestSyncSessionRateLimitPreservesBindingsAndRecovers(t *testing.T) {
a, s := testApp(t)
ctx := context.Background()
s, err := a.Mutate(ctx, s.Revision, func(d *domain.Dataset) error {
d.Accounts[0].ExternalAccountID = "main_uid"
for _, id := range []string{"mwst", "tax", "independent"} {
d.Accounts = append(d.Accounts, domain.Account{ID: id, DisplayName: id, Currency: "EUR", Active: true, ExternalAccountID: id + "_uid"})
}
return nil
})
if err != nil {
t.Fatal(err)
}
expiry := time.Now().Add(24 * time.Hour).Format(time.RFC3339)
limited := banking.Session{ID: "limited", ValidUntil: expiry}
healthy := banking.Session{ID: "healthy", ValidUntil: expiry}
for _, account := range s.Data.Accounts {
if account.ID == "independent" {
healthy.Accounts = append(healthy.Accounts, account)
} else {
limited.Accounts = append(limited.Accounts, account)
}
}
a.ops.Sessions = []banking.Session{limited, healthy}
a.ops.Consents["limited"] = Consent{Institution: "N26", Country: "DE", HistoryMonths: 24}
a.ops.Consents["healthy"] = Consent{Institution: "Other", Country: "DE", HistoryMonths: 12}
b := &sessionBank{
statuses: map[string]banking.SessionStatus{
"limited": {ValidUntil: expiry, AccountIDs: []string{"main_uid", "mwst_uid", "tax_uid"}},
// Another consent must not authorize an account whose own status failed.
"healthy": {ValidUntil: expiry, AccountIDs: []string{"independent_uid", "main_uid"}},
},
failures: map[string]error{},
}
a.bank = b
before, err := a.Sync(ctx)
if err != nil || len(before.Data.Transactions) != 4 {
t.Fatalf("initial sync: transactions=%d, error=%v, sync error=%s", len(before.Data.Transactions), err, before.Status.SyncError)
}
old := time.Now().Add(-48 * time.Hour).UTC().Format(time.RFC3339)
a.ops.LastSync = old
for _, account := range before.Data.Accounts {
a.ops.AccountSync[account.ID] = old
}
b.fetched = nil
b.failures["limited"] = bankRateError(t)
failed, err := a.Sync(ctx)
if err != nil {
t.Fatal(err)
}
if strings.Count(failed.Status.SyncError, "429") != 1 || strings.Contains(failed.Status.SyncError, "bank connection unavailable") || strings.Contains(failed.Status.SyncError, "private provider response") || !strings.Contains(failed.Status.SyncError, "retry") {
t.Fatal("session rate limit was duplicated, obscured, or exposed private data")
}
if !reflect.DeepEqual(b.fetched, []string{"independent"}) {
t.Fatal("failed consent authorized retrieval or independent consent stopped syncing")
}
if !reflect.DeepEqual(before.Data, failed.Data) || !reflect.DeepEqual(before.Sessions, failed.Sessions) || a.ops.LastSync != old {
t.Fatal("rate limit changed existing facts, bindings, metadata, or last successful sync")
}
for _, account := range before.Data.Accounts {
if account.ID != "independent" && a.ops.AccountSync[account.ID] != old {
t.Fatal("failed account advanced its cursor")
}
}
if a.ops.AccountSync["independent"] == old || a.ops.Consents["limited"].NeedsReconnect || a.ops.Consents["limited"].HistoryMonths != 24 {
t.Fatal("rate limit lost consent settings, required reconnect, or stopped the healthy cursor")
}
a = reopenBankingApp(t, a)
a.bank = b
if a.ops.Consents["limited"].NeedsReconnect || a.ops.LastSync != old || a.ops.AccountSync["mwst"] != old {
t.Fatal("rate failure state did not survive restart safely")
}
delete(b.failures, "limited")
b.fetched = nil
recovered, err := a.Sync(ctx)
if err != nil {
t.Fatal(err)
}
if recovered.Status.SyncError != "" || a.ops.Consents["limited"].Error != "" || a.ops.Consents["limited"].NeedsReconnect || a.ops.LastSync == old {
t.Fatal("successful retry did not clear the transient failure")
}
if !reflect.DeepEqual(before.Data, recovered.Data) || len(b.fetched) != 4 {
t.Fatal("recovery duplicated facts or skipped an account")
}
for _, account := range recovered.Data.Accounts {
if a.ops.AccountSync[account.ID] == old {
t.Fatal("recovered account cursor did not advance")
}
}
}
func TestSyncMissingMembershipStillRejectsAccount(t *testing.T) {
a, s, b := backfillApp(t)
b.session.Accounts = []domain.Account{s.Data.Accounts[1]}
before := domain.Clone(s.Data)
last := a.ops.LastSync
after, err := a.Sync(context.Background())
if err != nil {
t.Fatal(err)
}
if !strings.Contains(after.Status.SyncError, s.Data.Accounts[0].DisplayName+": bank connection unavailable") || !a.ops.Consents["current"].NeedsReconnect {
t.Fatal("missing account membership was treated as authorized")
}
if len(b.accounts) != 1 || b.accounts[0].ID != "other" || a.ops.AccountSync[s.Data.Accounts[0].ID] != last || a.ops.LastSync != last {
t.Fatal("missing member was fetched or advanced its cursor, or valid member was skipped")
}
if !reflect.DeepEqual(before.Accounts, after.Data.Accounts) || len(after.Data.Transactions) != 1 || after.Data.Transactions[0].Facts.AccountID != "other" {
t.Fatal("missing membership changed bindings or imported unauthorized facts")
}
}
func TestSyncTransactionFailuresPreserveProgressAndSafeErrors(t *testing.T) {
for _, scenario := range []string{"rate limit", "reconnect", "private response"} {
t.Run(scenario, func(t *testing.T) {
a, s, b := backfillApp(t)
s = seed(t, a, s)
old := a.ops.LastSync
cursors := map[string]string{}
for id, cursor := range a.ops.AccountSync {
cursors[id] = cursor
}
switch scenario {
case "rate limit":
b.fetchErr = bankRateError(t)
case "reconnect":
b.fetchErr = fmt.Errorf("private provider response: %w", banking.ErrReconnect)
case "private response":
b.fetchErr = errors.New("private provider response")
}
failed, err := a.Sync(context.Background())
if err != nil {
t.Fatal(err)
}
meta := a.ops.Consents["current"]
if failed.Status.SyncError == "" || meta.Error == "" || strings.Contains(failed.Status.SyncError+meta.Error, "private provider response") {
t.Fatal("transaction failure was lost or exposed provider data")
}
if scenario == "rate limit" && (!strings.Contains(failed.Status.SyncError, "429") || !strings.Contains(meta.Error, "429") || !strings.Contains(meta.Error, "retry")) {
t.Fatal("transaction rate error was obscured")
}
if meta.NeedsReconnect != (scenario == "reconnect") {
t.Fatal("transaction failure classified consent incorrectly")
}
if !reflect.DeepEqual(s.Data, failed.Data) || !reflect.DeepEqual(cursors, a.ops.AccountSync) || a.ops.LastSync != old {
t.Fatal("failed retrieval imported partial data or advanced synchronization")
}
b.fetchErr = nil
recovered, err := a.Sync(context.Background())
if err != nil || recovered.Status.SyncError != "" || a.ops.Consents["current"].Error != "" || a.ops.Consents["current"].NeedsReconnect {
t.Fatal("successful retrieval did not clear the failure")
}
})
}
}
+49 -43
View File
@@ -23,6 +23,7 @@ import (
"time" "time"
"finance-duck/internal/domain" "finance-duck/internal/domain"
"finance-duck/internal/ratelimit"
) )
// ErrReconnect identifies inactive bank consent, not application authentication // ErrReconnect identifies inactive bank consent, not application authentication
@@ -34,6 +35,13 @@ type Session struct {
ValidUntil string `json:"valid_until"` ValidUntil string `json:"valid_until"`
Accounts []domain.Account `json:"accounts"` Accounts []domain.Account `json:"accounts"`
} }
// SessionStatus contains only the current consent expiry and external account
// membership. Full account metadata is captured once by Exchange.
type SessionStatus struct {
ValidUntil string
AccountIDs []string
}
type Balance struct { type Balance struct {
Amount domain.Money `json:"amount"` Amount domain.Money `json:"amount"`
Currency string `json:"currency"` Currency string `json:"currency"`
@@ -43,7 +51,7 @@ type Balance struct {
type Provider interface { type Provider interface {
Authorize(context.Context, string, string, string) (string, error) Authorize(context.Context, string, string, string) (string, error)
Exchange(context.Context, string) (Session, error) Exchange(context.Context, string) (Session, error)
Status(context.Context, string) (Session, error) Status(context.Context, string) (SessionStatus, error)
Balances(context.Context, string) ([]Balance, error) Balances(context.Context, string) ([]Balance, error)
Transactions(context.Context, domain.Account, string, string) ([]domain.Facts, error) Transactions(context.Context, domain.Account, string, string) ([]domain.Facts, error)
} }
@@ -53,6 +61,7 @@ type EnableBanking struct {
appID string appID string
key *rsa.PrivateKey key *rsa.PrivateKey
redirectURL string redirectURL string
requests ratelimit.Controller
} }
var _ Provider = (*EnableBanking)(nil) var _ Provider = (*EnableBanking)(nil)
@@ -125,9 +134,10 @@ func (p *EnableBanking) jwt() (string, error) {
return unsigned + "." + base64.RawURLEncoding.EncodeToString(signature), nil return unsigned + "." + base64.RawURLEncoding.EncodeToString(signature), nil
} }
func (p *EnableBanking) request(ctx context.Context, method, path string, input, output any) error { func (p *EnableBanking) request(ctx context.Context, method, path string, input, output any) error {
// Enforce a deadline even when a caller injects a client without Timeout. if err := p.requests.Acquire(ctx); err != nil {
ctx, cancel := context.WithTimeout(ctx, 30*time.Second) return fmt.Errorf("Enable Banking: %w", err)
defer cancel() }
defer p.requests.Release()
token, err := p.jwt() token, err := p.jwt()
if err != nil { if err != nil {
return err return err
@@ -157,17 +167,27 @@ func (p *EnableBanking) request(ctx context.Context, method, path string, input,
if p.HTTPClient != nil { if p.HTTPClient != nil {
client = *p.HTTPClient client = *p.HTTPClient
} }
// Cap each attempt, including response reads, without timing out retry waits.
if client.Timeout <= 0 || client.Timeout > 30*time.Second {
client.Timeout = 30 * time.Second
}
// Never forward signed credentials or financial requests through redirects. // Never forward signed credentials or financial requests through redirects.
client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
response, err := client.Do(req) response, err := p.requests.Do(ctx, func(ctx context.Context) (*http.Response, error) {
response, err := client.Do(req.Clone(ctx))
if err != nil {
if errors.Is(err, context.Canceled) {
return nil, context.Canceled
}
if errors.Is(err, context.DeadlineExceeded) {
return nil, fmt.Errorf("request timed out: %w", context.DeadlineExceeded)
}
return nil, errors.New("connection failed")
}
return response, nil
}, method == http.MethodGet)
if err != nil { if err != nil {
if errors.Is(err, context.Canceled) { return fmt.Errorf("Enable Banking: %w", err)
return context.Canceled
}
if errors.Is(err, context.DeadlineExceeded) {
return fmt.Errorf("Enable Banking request timed out")
}
return fmt.Errorf("Enable Banking connection failed")
} }
defer response.Body.Close() defer response.Body.Close()
if response.StatusCode < 200 || response.StatusCode >= 300 { if response.StatusCode < 200 || response.StatusCode >= 300 {
@@ -176,6 +196,12 @@ func (p *EnableBanking) request(ctx context.Context, method, path string, input,
const limit = 16 << 20 const limit = 16 << 20
b, err := io.ReadAll(io.LimitReader(response.Body, limit+1)) b, err := io.ReadAll(io.LimitReader(response.Body, limit+1))
if err != nil { if err != nil {
if errors.Is(err, context.Canceled) {
return context.Canceled
}
if errors.Is(err, context.DeadlineExceeded) {
return fmt.Errorf("Enable Banking response timed out: %w", context.DeadlineExceeded)
}
return fmt.Errorf("read Enable Banking response") return fmt.Errorf("read Enable Banking response")
} }
if len(b) > limit { if len(b) > limit {
@@ -311,54 +337,34 @@ func (p *EnableBanking) Exchange(ctx context.Context, code string) (Session, err
} }
return result, nil return result, nil
} }
func (p *EnableBanking) Status(ctx context.Context, sessionID string) (Session, error) { func (p *EnableBanking) Status(ctx context.Context, sessionID string) (SessionStatus, error) {
if sessionID == "" { if sessionID == "" {
return Session{}, fmt.Errorf("session ID is required") return SessionStatus{}, fmt.Errorf("session ID is required")
} }
var response struct { var response struct {
Status string `json:"status"` Status string `json:"status"`
Accounts []string `json:"accounts"` Accounts []string `json:"accounts"`
AccountsData []accountDTO `json:"accounts_data"` Access accessDTO `json:"access"`
Access accessDTO `json:"access"`
ASPSP institutionDTO `json:"aspsp"`
} }
if err := p.request(ctx, http.MethodGet, "/sessions/"+url.PathEscape(sessionID), nil, &response); err != nil { if err := p.request(ctx, http.MethodGet, "/sessions/"+url.PathEscape(sessionID), nil, &response); err != nil {
return Session{}, err return SessionStatus{}, err
} }
if response.Status != "AUTHORIZED" { if response.Status != "AUTHORIZED" {
return Session{}, fmt.Errorf("Enable Banking session is not authorized: %w", ErrReconnect) return SessionStatus{}, fmt.Errorf("Enable Banking session is not authorized: %w", ErrReconnect)
} }
expires, err := time.Parse(time.RFC3339, response.Access.ValidUntil) expires, err := time.Parse(time.RFC3339, response.Access.ValidUntil)
if err != nil { if err != nil {
return Session{}, fmt.Errorf("Enable Banking returned invalid session expiry") return SessionStatus{}, fmt.Errorf("Enable Banking returned invalid session expiry")
} }
if !expires.After(time.Now()) { if !expires.After(time.Now()) {
return Session{}, fmt.Errorf("Enable Banking session expired: %w", ErrReconnect) return SessionStatus{}, fmt.Errorf("Enable Banking session expired: %w", ErrReconnect)
}
result := Session{ID: sessionID, ValidUntil: response.Access.ValidUntil, Accounts: []domain.Account{}}
hashes := map[string]string{}
for _, a := range response.AccountsData {
hashes[a.UID] = a.IdentificationHash
} }
for _, id := range response.Accounts { for _, id := range response.Accounts {
if id == "" { if id == "" {
return Session{}, fmt.Errorf("Enable Banking returned empty account identifier") return SessionStatus{}, fmt.Errorf("Enable Banking returned empty account identifier")
} }
var details accountDTO
if err := p.request(ctx, http.MethodGet, "/accounts/"+url.PathEscape(id)+"/details", nil, &details); err != nil {
return Session{}, err
}
details.UID = id
if details.IdentificationHash == "" {
details.IdentificationHash = hashes[id]
}
a, err := details.account(response.ASPSP.Name)
if err != nil {
return Session{}, err
}
result.Accounts = append(result.Accounts, a)
} }
return result, nil return SessionStatus{ValidUntil: response.Access.ValidUntil, AccountIDs: response.Accounts}, nil
} }
type amountDTO struct { type amountDTO struct {
+200 -3
View File
@@ -13,11 +13,15 @@ import (
"encoding/pem" "encoding/pem"
"errors" "errors"
"fmt" "fmt"
"io"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strings" "strings"
"sync/atomic"
"testing" "testing"
"time" "time"
"finance-duck/internal/ratelimit"
) )
func testProvider(t *testing.T, handler http.HandlerFunc) (*EnableBanking, *rsa.PrivateKey) { func testProvider(t *testing.T, handler http.HandlerFunc) (*EnableBanking, *rsa.PrivateKey) {
@@ -138,7 +142,8 @@ func TestEnableBankingDocumentedFlowAndPagination(t *testing.T) {
case "/sessions/session-1": case "/sessions/session-1":
fmt.Fprintf(w, `{"status":"AUTHORIZED","access":{"valid_until":%q},"aspsp":{"name":"N26","country":"DE"},"accounts":["uid-one"],"accounts_data":[{"uid":"uid-one","identification_hash":"stable-hash"}]}`, expiry) fmt.Fprintf(w, `{"status":"AUTHORIZED","access":{"valid_until":%q},"aspsp":{"name":"N26","country":"DE"},"accounts":["uid-one"],"accounts_data":[{"uid":"uid-one","identification_hash":"stable-hash"}]}`, expiry)
case "/accounts/uid-one/details": case "/accounts/uid-one/details":
fmt.Fprint(w, `{"account_id":{"iban":"DE02120300000000202051"},"details":"Main account","currency":"EUR"}`) t.Error("session membership must not require account details")
http.Error(w, "account details unavailable", http.StatusServiceUnavailable)
case "/accounts/uid-one/balances": case "/accounts/uid-one/balances":
fmt.Fprint(w, `{"balances":[{"name":"Booked","balance_amount":{"currency":"EUR","amount":"1234.5678"},"balance_type":"CLBD","reference_date":"2026-09-01"}]}`) fmt.Fprint(w, `{"balances":[{"name":"Booked","balance_amount":{"currency":"EUR","amount":"1234.5678"},"balance_type":"CLBD","reference_date":"2026-09-01"}]}`)
case "/accounts/uid-one/transactions": case "/accounts/uid-one/transactions":
@@ -180,8 +185,8 @@ func TestEnableBankingDocumentedFlowAndPagination(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(status.Accounts) != 1 || status.Accounts[0].ID != session.Accounts[0].ID || status.Accounts[0].ExternalAccountID != "uid-one" { if len(status.AccountIDs) != 1 || status.AccountIDs[0] != session.Accounts[0].ExternalAccountID || status.ValidUntil != expiry {
t.Fatalf("account identity changed between session DTOs: %+v", status) t.Fatalf("incorrect consent membership or expiry: %+v", status)
} }
balances, err := p.Balances(context.Background(), "uid-one") balances, err := p.Balances(context.Background(), "uid-one")
if err != nil || len(balances) != 1 || balances[0].Amount.String() != "1234.5678" || balances[0].Type != "CLBD" { if err != nil || len(balances) != 1 || balances[0].Amount.String() != "1234.5678" || balances[0].Type != "CLBD" {
@@ -265,6 +270,198 @@ func TestEnableBankingExpiredConsentRequiresReconnect(t *testing.T) {
} }
} }
func TestEnableBankingRejectsInvalidSessionMembership(t *testing.T) {
for name, payload := range map[string]string{
"empty UID": `{"status":"AUTHORIZED","accounts":[""],"access":{"valid_until":"2099-01-01T00:00:00Z"}}`,
"non-string UID": `{"status":"AUTHORIZED","accounts":[{}],"access":{"valid_until":"2099-01-01T00:00:00Z"}}`,
"invalid expiry": `{"status":"AUTHORIZED","accounts":["uid"],"access":{"valid_until":"not-a-date"}}`,
} {
t.Run(name, func(t *testing.T) {
p, _ := testProvider(t, func(w http.ResponseWriter, r *http.Request) {
fmt.Fprint(w, payload)
})
status, err := p.Status(context.Background(), "session")
if err == nil || status.ValidUntil != "" || status.AccountIDs != nil || errors.Is(err, ErrReconnect) {
t.Fatalf("invalid response returned usable membership or claimed revoked consent: %+v %v", status, err)
}
})
}
}
type bankingRoundTripFunc func(*http.Request) (*http.Response, error)
func (f bankingRoundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) {
return f(r)
}
type bankingClosedBody struct {
io.ReadCloser
closed *atomic.Int32
onClose func()
}
func (b *bankingClosedBody) Close() error {
b.closed.Add(1)
err := b.ReadCloser.Close()
if b.onClose != nil {
b.onClose()
}
return err
}
func TestEnableBankingGETRecoversAfterRateLimit(t *testing.T) {
for _, endpoint := range []string{"status", "transactions"} {
t.Run(endpoint, func(t *testing.T) {
t.Parallel()
var calls, closed atomic.Int32
var first atomic.Int64
p, _ := testProvider(t, func(w http.ResponseWriter, r *http.Request) {
if calls.Add(1) == 1 {
first.Store(time.Now().UnixNano())
w.Header().Set("Retry-After", "1")
http.Error(w, "private provider response", http.StatusTooManyRequests)
return
}
if time.Since(time.Unix(0, first.Load())) < time.Second {
t.Error("retried before provider cooldown elapsed")
}
if closed.Load() != 1 {
t.Error("retried without closing the rate-limit response")
}
if endpoint == "status" {
fmt.Fprint(w, `{"status":"AUTHORIZED","accounts":["uid"],"access":{"valid_until":"2099-01-01T00:00:00Z"}}`)
} else {
fmt.Fprint(w, `{"transactions":[{"entry_reference":"entry","transaction_amount":{"amount":"1.00","currency":"EUR"},"credit_debit_indicator":"CRDT","status":"BOOK","booking_date":"2026-09-01"}]}`)
}
})
transport := p.HTTPClient.Transport
p.HTTPClient.Timeout = 500 * time.Millisecond
p.HTTPClient.Transport = bankingRoundTripFunc(func(r *http.Request) (*http.Response, error) {
response, err := transport.RoundTrip(r)
if err == nil && response.StatusCode == http.StatusTooManyRequests {
response.Body = &bankingClosedBody{ReadCloser: response.Body, closed: &closed}
}
return response, err
})
if endpoint == "status" {
status, err := p.Status(context.Background(), "session")
if err != nil || len(status.AccountIDs) != 1 || status.AccountIDs[0] != "uid" {
t.Fatalf("session membership did not recover: %+v %v", status, err)
}
} else {
account := fixtureDataset().Accounts[0]
account.ExternalAccountID = "uid"
rows, err := p.Transactions(context.Background(), account, "", "")
if err != nil || len(rows) != 1 || rows[0].ExternalID != "entry" || rows[0].Amount.String() != "1.00" {
t.Fatalf("transaction retrieval did not recover: %+v %v", rows, err)
}
}
if calls.Load() != 2 || closed.Load() != 1 {
t.Fatalf("unexpected retry requests or response leaks: calls=%d closed=%d", calls.Load(), closed.Load())
}
})
}
}
func TestEnableBankingCooldownCoversAllEndpoints(t *testing.T) {
var calls atomic.Int32
p, _ := testProvider(t, func(w http.ResponseWriter, r *http.Request) {
calls.Add(1)
w.Header().Set("Retry-After", "300")
http.Error(w, "private provider response", http.StatusTooManyRequests)
})
_, err := p.Status(context.Background(), "session")
var initial *ratelimit.RateLimitError
if !errors.As(err, &initial) || !initial.RetryAt().After(time.Now()) || errors.Is(err, ErrReconnect) || strings.Contains(err.Error(), "private") {
t.Fatalf("unsafe or missing rate-limit error: %v", err)
}
account := fixtureDataset().Accounts[0]
account.ExternalAccountID = "uid"
for name, request := range map[string]func() error{
"status": func() error { _, err := p.Status(context.Background(), "other-session"); return err },
"balances": func() error { _, err := p.Balances(context.Background(), "uid"); return err },
"transactions": func() error { _, err := p.Transactions(context.Background(), account, "", ""); return err },
"exchange": func() error { _, err := p.Exchange(context.Background(), "once-only-code"); return err },
"authorize": func() error { _, err := p.Authorize(context.Background(), "N26", "DE", "state"); return err },
} {
t.Run(name, func(t *testing.T) {
err := request()
var limit *ratelimit.RateLimitError
if !errors.As(err, &limit) || !limit.RetryAt().Equal(initial.RetryAt()) || errors.Is(err, ErrReconnect) || strings.Contains(err.Error(), "private") {
t.Fatalf("cooldown was lost or unsafe: %v", err)
}
if calls.Load() != 1 {
t.Fatalf("provider contacted during cooldown: %d requests", calls.Load())
}
})
}
}
func TestEnableBankingNeverReplaysMutationAfterRateLimit(t *testing.T) {
for _, endpoint := range []string{"exchange", "authorize"} {
t.Run(endpoint, func(t *testing.T) {
var posts atomic.Int32
p, _ := testProvider(t, func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/aspsps" {
fmt.Fprint(w, `{"aspsps":[{"name":"N26","country":"DE","maximum_consent_validity":3600}]}`)
return
}
if r.Method != http.MethodPost {
t.Error("unexpected non-mutation request")
}
posts.Add(1)
w.Header().Set("Retry-After", "1")
http.Error(w, "private once-only exchange failure", http.StatusTooManyRequests)
})
var err error
if endpoint == "exchange" {
_, err = p.Exchange(context.Background(), "once-only-code")
} else {
_, err = p.Authorize(context.Background(), "N26", "DE", "state")
}
var limit *ratelimit.RateLimitError
if !errors.As(err, &limit) || strings.Contains(err.Error(), "private") || errors.Is(err, ErrReconnect) {
t.Fatalf("mutation rate limit was lost or unsafe: %v", err)
}
if posts.Load() != 1 {
t.Fatalf("mutation replayed %d times", posts.Load())
}
})
}
}
func TestEnableBankingCanceledRetryRetainsCooldown(t *testing.T) {
var calls, closed atomic.Int32
p, _ := testProvider(t, func(w http.ResponseWriter, r *http.Request) {
calls.Add(1)
w.Header().Set("Retry-After", "60")
http.Error(w, "private provider response", http.StatusTooManyRequests)
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
transport := p.HTTPClient.Transport
p.HTTPClient.Transport = bankingRoundTripFunc(func(r *http.Request) (*http.Response, error) {
response, err := transport.RoundTrip(r)
if err == nil && response.StatusCode == http.StatusTooManyRequests {
response.Body = &bankingClosedBody{ReadCloser: response.Body, closed: &closed, onClose: cancel}
}
return response, err
})
_, err := p.Status(ctx, "session")
var original *ratelimit.RateLimitError
if !errors.Is(err, context.Canceled) || !errors.As(err, &original) || strings.Contains(err.Error(), "private") {
t.Fatalf("cancellation lost safe rate-limit evidence: %v", err)
}
_, err = p.Balances(context.Background(), "uid")
var retained *ratelimit.RateLimitError
if !errors.As(err, &retained) || !retained.RetryAt().Equal(original.RetryAt()) {
t.Fatalf("cancellation discarded provider cooldown: %v", err)
}
if calls.Load() != 1 || closed.Load() != 1 {
t.Fatalf("cancellation retried or leaked a response: calls=%d closed=%d", calls.Load(), closed.Load())
}
}
func TestEnableBankingValidatesUploadedCredentials(t *testing.T) { func TestEnableBankingValidatesUploadedCredentials(t *testing.T) {
_, key := testProvider(t, func(w http.ResponseWriter, r *http.Request) { _, key := testProvider(t, func(w http.ResponseWriter, r *http.Request) {
t.Error("credential validation must not call provider") t.Error("credential validation must not call provider")
+85 -17
View File
@@ -11,18 +11,63 @@ import (
"net/http" "net/http"
"net/url" "net/url"
"strings" "strings"
"sync/atomic"
"time" "time"
"unicode/utf8" "unicode/utf8"
"finance-duck/internal/domain" "finance-duck/internal/domain"
"finance-duck/internal/ratelimit"
) )
// Client configuration must not be mutated concurrently with classification.
// Do not copy a Client after use; use WithModel to share its rate control safely.
type Client struct { type Client struct {
APIKey string APIKey string
Model string Model string
IncludeAmount bool IncludeAmount bool
HTTPClient *http.Client HTTPClient *http.Client
BaseURL string BaseURL string
rate atomic.Pointer[ratelimit.Controller]
}
// WithModel snapshots the configuration while sharing the original client's
// in-flight request gate and provider cooldown, including across model choices.
func (c *Client) WithModel(model string) *Client {
snapshot := &Client{
APIKey: c.APIKey,
Model: model,
IncludeAmount: c.IncludeAmount,
HTTPClient: c.HTTPClient,
BaseURL: c.BaseURL,
}
snapshot.rate.Store(c.rateControl())
return snapshot
}
func (c *Client) rateControl() *ratelimit.Controller {
if gate := c.rate.Load(); gate != nil {
return gate
}
gate := &ratelimit.Controller{}
if c.rate.CompareAndSwap(nil, gate) {
return gate
}
return c.rate.Load()
}
// Keep context identity without exposing transport errors containing URLs or
// response details, including deadlines enforced by http.Client itself.
func requestContextError(ctx context.Context, err error) error {
if cause := ctx.Err(); cause != nil {
return cause
}
for _, cause := range []error{context.Canceled, context.DeadlineExceeded} {
if errors.Is(err, cause) {
return cause
}
}
return nil
} }
type Proposal struct { type Proposal struct {
@@ -41,9 +86,12 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
} }
} }
p := Proposal{Enrichment: domain.Fallback(facts)} p := Proposal{Enrichment: domain.Fallback(facts)}
failError := func(err error) (Proposal, error) {
p.Enrichment.Classification = domain.Provenance{Source: "fallback", Timestamp: time.Now().UTC().Format(time.RFC3339), Error: err.Error()}
return p, err
}
fail := func(message string) (Proposal, error) { fail := func(message string) (Proposal, error) {
p.Enrichment.Classification = domain.Provenance{Source: "fallback", Timestamp: time.Now().UTC().Format(time.RFC3339), Error: message} return failError(errors.New(message))
return p, errors.New(message)
} }
if _, err := facts.Amount.Minor(); err != nil { if _, err := facts.Amount.Minor(); err != nil {
return fail("invalid transaction amount") return fail("invalid transaction amount")
@@ -64,9 +112,16 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
return p, nil return p, nil
} }
} }
if strings.TrimSpace(c.APIKey) == "" || strings.TrimSpace(c.Model) == "" { apiKey, model := c.APIKey, c.Model
includeAmount, baseURL, configuredHTTPClient := c.IncludeAmount, c.BaseURL, c.HTTPClient
if strings.TrimSpace(apiKey) == "" || strings.TrimSpace(model) == "" {
return fail("AI classification is not configured") return fail("AI classification is not configured")
} }
gate := c.rateControl()
if err := gate.Acquire(ctx); err != nil {
return failError(err)
}
defer gate.Release()
clean := newSanitizer(facts, data, false) clean := newSanitizer(facts, data, false)
merchantClean := newSanitizer(facts, data, true) merchantClean := newSanitizer(facts, data, true)
candidates := retrieve(localDescription, p.Enrichment.Kind, data, clean, merchantClean) candidates := retrieve(localDescription, p.Enrichment.Kind, data, clean, merchantClean)
@@ -78,7 +133,7 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
Amount *domain.Money `json:"amount,omitempty"` Amount *domain.Money `json:"amount,omitempty"`
Currency string `json:"currency,omitempty"` Currency string `json:"currency,omitempty"`
}{Description: clean(facts.RawDescription), Categories: candidates.categories, Tags: candidates.tags, Merchants: candidates.merchants} }{Description: clean(facts.RawDescription), Categories: candidates.categories, Tags: candidates.tags, Merchants: candidates.merchants}
if c.IncludeAmount { if includeAmount {
prompt.Amount = &facts.Amount prompt.Amount = &facts.Amount
// Currency is validated separately rather than copied from arbitrary bank text. // Currency is validated separately rather than copied from arbitrary bank text.
if len(facts.Currency) != 3 || strings.IndexFunc(facts.Currency, func(r rune) bool { return r < 'A' || r > 'Z' }) >= 0 { if len(facts.Currency) != 3 || strings.IndexFunc(facts.Currency, func(r rune) bool { return r < 'A' || r > 'Z' }) >= 0 {
@@ -91,7 +146,7 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
return fail("cannot encode classification request") return fail("cannot encode classification request")
} }
request := map[string]any{ request := map[string]any{
"model": c.Model, "model": model,
"stream": false, "stream": false,
"max_tokens": 512, "max_tokens": 512,
// Fail closed: never retry without these controls. No plugins/tools are enabled. // Fail closed: never retry without these controls. No plugins/tools are enabled.
@@ -108,7 +163,7 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
if err != nil { if err != nil {
return fail("cannot encode classification request") return fail("cannot encode classification request")
} }
base := strings.TrimRight(c.BaseURL, "/") base := strings.TrimRight(baseURL, "/")
if base == "" { if base == "" {
base = "https://openrouter.ai/api/v1" base = "https://openrouter.ai/api/v1"
} }
@@ -119,24 +174,34 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
if endpoint.Scheme != "https" && !(endpoint.Scheme == "http" && (endpoint.Hostname() == "localhost" || endpoint.Hostname() == "127.0.0.1" || endpoint.Hostname() == "::1")) { if endpoint.Scheme != "https" && !(endpoint.Scheme == "http" && (endpoint.Hostname() == "localhost" || endpoint.Hostname() == "127.0.0.1" || endpoint.Hostname() == "::1")) {
return fail("AI endpoint must use HTTPS") return fail("AI endpoint must use HTTPS")
} }
req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/chat/completions", bytes.NewReader(body))
if err != nil {
return fail("cannot create classification request")
}
req.Header.Set("Authorization", "Bearer "+c.APIKey)
req.Header.Set("Content-Type", "application/json")
client := http.Client{Timeout: 45 * time.Second} client := http.Client{Timeout: 45 * time.Second}
if c.HTTPClient != nil { if configuredHTTPClient != nil {
client = *c.HTTPClient client = *configuredHTTPClient
if client.Timeout == 0 { if client.Timeout == 0 {
client.Timeout = 45 * time.Second client.Timeout = 45 * time.Second
} }
} }
// Redirects could send sensitive prompts to endpoints with different policies. // Redirects could send sensitive prompts to endpoints with different policies.
client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
resp, err := client.Do(req) resp, err := gate.Do(ctx, func(ctx context.Context) (*http.Response, error) {
// Each attempt uses identical serialized bytes, credentials and controls.
req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/chat/completions", bytes.NewReader(body))
if err != nil {
return nil, errors.New("cannot create classification request")
}
req.Header.Set("Authorization", "Bearer "+apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
if cause := requestContextError(ctx, err); cause != nil {
return nil, fmt.Errorf("AI request canceled: %w", cause)
}
return nil, errors.New("AI request failed")
}
return resp, nil
}, true)
if err != nil { if err != nil {
return fail("AI request failed") return failError(err)
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
@@ -145,6 +210,9 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
const maxResponse = 64 * 1024 const maxResponse = 64 * 1024
raw, err := io.ReadAll(io.LimitReader(resp.Body, maxResponse+1)) raw, err := io.ReadAll(io.LimitReader(resp.Body, maxResponse+1))
if err != nil || len(raw) > maxResponse { if err != nil || len(raw) > maxResponse {
if cause := requestContextError(ctx, err); cause != nil {
return failError(fmt.Errorf("AI request canceled: %w", cause))
}
return fail("invalid AI response size") return fail("invalid AI response size")
} }
var envelope struct { var envelope struct {
@@ -202,7 +270,7 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D
e.MerchantID = proposed.ID e.MerchantID = proposed.ID
} }
} }
e.Classification = domain.Provenance{Source: "openrouter", Model: c.Model, Timestamp: time.Now().UTC().Format(time.RFC3339)} e.Classification = domain.Provenance{Source: "openrouter", Model: model, Timestamp: time.Now().UTC().Format(time.RFC3339)}
validationData := data validationData := data
if proposed != nil { if proposed != nil {
validationData.Merchants = append(append([]domain.Merchant{}, data.Merchants...), *proposed) validationData.Merchants = append(append([]domain.Merchant{}, data.Merchants...), *proposed)
+1 -1
View File
@@ -290,7 +290,7 @@ func TestUnsafeMerchantProposalRejected(t *testing.T) {
} }
func TestProviderErrorsNeverRelaxPolicyOrEchoResponse(t *testing.T) { func TestProviderErrorsNeverRelaxPolicyOrEchoResponse(t *testing.T) {
for _, status := range []int{302, 400, 401, 404, 429, 500, 503} { for _, status := range []int{302, 400, 401, 402, 403, 404, 500, 503} {
t.Run(fmt.Sprint(status), func(t *testing.T) { t.Run(fmt.Sprint(status), func(t *testing.T) {
f, d := fixture() f, d := fixture()
calls := 0 calls := 0
+341
View File
@@ -0,0 +1,341 @@
package classification
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"net/http"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
)
func TestRateLimitRetryPreservesPrivateRequest(t *testing.T) {
f, d := fixture()
f.Counterparty = "Alice Privateperson"
f.CounterpartyIBAN = "DE89370400440532013000"
f.RawDescription = "Coffee House Alice Privateperson DE89370400440532013000 private_external -918.27 reference secretpayment"
var requests [][]byte
var arrivals []time.Time
var mu sync.Mutex
c := mockClient(t, func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
arrivals = append(arrivals, time.Now())
body, err := io.ReadAll(r.Body)
if err != nil {
t.Error(err)
}
requests = append(requests, body)
if r.Method != http.MethodPost || r.URL.Path != "/chat/completions" || r.Header.Get("Authorization") != "Bearer test-secret" || r.Header.Get("Content-Type") != "application/json" {
t.Error("retry changed authenticated JSON endpoint")
}
if len(requests) == 1 {
w.Header().Set("Retry-After", "2")
w.WriteHeader(http.StatusTooManyRequests)
_, _ = io.WriteString(w, "sensitive-provider-response")
return
}
reply(w, validAnswer)
})
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
defer cancel()
p, err := c.Classify(ctx, f, d, true)
if err != nil || p.Enrichment.Classification.Source != "openrouter" || p.Enrichment.Classification.Error != "" {
t.Fatalf("retry did not recover: %+v, %v", p, err)
}
mu.Lock()
defer mu.Unlock()
if len(requests) != 2 || !bytes.Equal(requests[0], requests[1]) {
t.Fatalf("retry did not reuse identical serialized request: %d attempts", len(requests))
}
if gap := arrivals[1].Sub(arrivals[0]); gap < 1950*time.Millisecond {
t.Fatalf("retried before provider's two-second delay: %v", gap)
}
var request struct {
Provider struct {
DataCollection string `json:"data_collection"`
ZDR bool `json:"zdr"`
Require bool `json:"require_parameters"`
} `json:"provider"`
Messages []struct{ Role, Content string } `json:"messages"`
ResponseFormat struct {
Type string `json:"type"`
Schema struct {
Strict bool `json:"strict"`
} `json:"json_schema"`
} `json:"response_format"`
Plugins json.RawMessage `json:"plugins"`
}
if err := json.Unmarshal(requests[0], &request); err != nil {
t.Fatal(err)
}
if request.Provider.DataCollection != "deny" || !request.Provider.ZDR || !request.Provider.Require || request.ResponseFormat.Type != "json_schema" || !request.ResponseFormat.Schema.Strict || len(request.Plugins) != 0 {
t.Fatal("retry relaxed private structured routing")
}
if len(request.Messages) != 2 {
t.Fatalf("unexpected message count: %d", len(request.Messages))
}
for _, secret := range []string{"alice", "privateperson", "3704", "private_external", "918", "secretpayment", "tx_private", "account_private"} {
if strings.Contains(strings.ToLower(request.Messages[1].Content), secret) {
t.Errorf("retried prompt leaked %q", secret)
}
}
}
func TestRateLimitBackoffExhaustionRetainsSharedCooldown(t *testing.T) {
f, d := fixture()
var arrivals []time.Time
var mu sync.Mutex
c := mockClient(t, func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
arrivals = append(arrivals, time.Now())
// An invalid hint must not bypass exponential fallback delays.
w.Header().Set("Retry-After", "not-a-delay")
w.WriteHeader(http.StatusTooManyRequests)
_, _ = io.WriteString(w, "sensitive-provider-response")
})
snapshot := c.WithModel("test/preview-model")
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
p, err := c.Classify(ctx, f, d, true)
assertSafeRateLimit(t, err, p.Enrichment.Classification.Error)
mu.Lock()
observed := append([]time.Time(nil), arrivals...)
mu.Unlock()
if len(observed) != 4 {
t.Fatalf("expected four bounded attempts, got %d", len(observed))
}
for i, delay := range []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} {
if gap := observed[i+1].Sub(observed[i]); gap < delay-50*time.Millisecond {
t.Errorf("backoff %d retried too early: %v, need %v", i+1, gap, delay)
}
}
for _, client := range []*Client{snapshot, c} {
probeCtx, probeCancel := context.WithTimeout(context.Background(), time.Second)
p, err := client.Classify(probeCtx, f, d, true)
probeCancel()
assertSafeRateLimit(t, err, p.Enrichment.Classification.Error)
mu.Lock()
calls := len(arrivals)
mu.Unlock()
if errors.Is(err, context.DeadlineExceeded) || calls != 4 {
t.Fatalf("retained cooldown waited or contacted provider: %v, attempts=%d", err, calls)
}
}
}
func TestRateLimitHTTPDateDoesNotRetryEarly(t *testing.T) {
f, d := fixture()
var retryAt time.Time
var arrivals []time.Time
var mu sync.Mutex
c := mockClient(t, func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
arrivals = append(arrivals, time.Now())
if len(arrivals) == 1 {
retryAt = time.Now().UTC().Add(3 * time.Second).Truncate(time.Second)
w.Header().Set("Retry-After", retryAt.Format(http.TimeFormat))
w.WriteHeader(http.StatusTooManyRequests)
return
}
reply(w, validAnswer)
})
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
defer cancel()
if p, err := c.Classify(ctx, f, d, true); err != nil || p.Enrichment.Classification.Source != "openrouter" {
t.Fatalf("HTTP-date retry failed: %+v, %v", p, err)
}
mu.Lock()
defer mu.Unlock()
if len(arrivals) != 2 {
t.Fatalf("expected one HTTP-date retry, got %d attempts", len(arrivals))
}
if arrivals[1].Before(retryAt.Add(-25 * time.Millisecond)) {
t.Fatalf("retried at %v before HTTP-date %v", arrivals[1], retryAt)
}
}
func TestRateLimitLongHintsFailFastAndLocalRulesBypassCooldown(t *testing.T) {
for _, hint := range []string{"600", time.Now().UTC().Add(10 * time.Minute).Format(http.TimeFormat), "9223372036854775807"} {
t.Run(hint, func(t *testing.T) {
f, d := fixture()
var calls atomic.Int32
c := mockClient(t, func(w http.ResponseWriter, r *http.Request) {
calls.Add(1)
w.Header().Set("Retry-After", hint)
w.WriteHeader(http.StatusTooManyRequests)
_, _ = io.WriteString(w, "sensitive-provider-response")
})
snapshot := c.WithModel("test/preview-model")
for _, client := range []*Client{c, snapshot} {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
p, err := client.Classify(ctx, f, d, true)
cancel()
assertSafeRateLimit(t, err, p.Enrichment.Classification.Error)
if errors.Is(err, context.DeadlineExceeded) || calls.Load() != 1 {
t.Fatalf("long hint waited or permitted an early request: %v, attempts=%d", err, calls.Load())
}
}
d.Merchants[0].UseDefaults = true
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
p, err := snapshot.Classify(ctx, f, d, false)
if err != nil || p.Enrichment.Classification.Source != "rule" || p.Enrichment.MerchantID != "mer_coffee" || calls.Load() != 1 {
t.Fatalf("cooldown blocked local rule: %+v, %v, attempts=%d", p, err, calls.Load())
}
})
}
}
func TestRateLimitCancellationClosesBodyAndRetainsCooldown(t *testing.T) {
f, d := fixture()
var calls atomic.Int32
lateRequest := make(chan struct{}, 4)
c := mockClient(t, func(w http.ResponseWriter, r *http.Request) {
if calls.Add(1) > 1 {
lateRequest <- struct{}{}
}
w.Header().Set("Retry-After", "1")
w.WriteHeader(http.StatusTooManyRequests)
_, _ = io.WriteString(w, "sensitive-provider-response")
})
closed := make(chan struct{})
var closeOnce sync.Once
transport := c.HTTPClient.Transport
c.HTTPClient.Transport = rateLimitRoundTripFunc(func(r *http.Request) (*http.Response, error) {
response, err := transport.RoundTrip(r)
if err == nil && response.StatusCode == http.StatusTooManyRequests {
response.Body = &rateLimitNotifyingBody{ReadCloser: response.Body, closed: closed, once: &closeOnce}
}
return response, err
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() {
_, err := c.Classify(ctx, f, d, true)
done <- err
}()
select {
case <-closed:
case <-time.After(2 * time.Second):
t.Fatal("429 response body was not closed before retry waiting")
}
cancel()
select {
case err := <-done:
if !errors.Is(err, context.Canceled) || strings.Contains(err.Error(), "sensitive") {
t.Fatalf("waiting cancellation did not preserve safe context identity: %v", err)
}
case <-time.After(time.Second):
t.Fatal("retry wait ignored cancellation")
}
probeCtx, probeCancel := context.WithTimeout(context.Background(), time.Second)
defer probeCancel()
p, err := c.WithModel("test/preview-model").Classify(probeCtx, f, d, true)
assertSafeRateLimit(t, err, p.Enrichment.Classification.Error)
if errors.Is(err, context.DeadlineExceeded) || calls.Load() != 1 {
t.Fatalf("cancelled retry lost cooldown or made a late request: %v, attempts=%d", err, calls.Load())
}
select {
case <-lateRequest:
t.Fatal("cancelled retry made a request after its timer expired")
case <-time.After(1200 * time.Millisecond):
}
}
func TestRateLimitQueuedSnapshotCancellationMakesNoLateRequest(t *testing.T) {
f, d := fixture()
entered := make(chan struct{})
release := make(chan struct{})
var releaseOnce sync.Once
unblock := func() { releaseOnce.Do(func() { close(release) }) }
var calls atomic.Int32
models := make(chan string, 4)
c := mockClient(t, func(w http.ResponseWriter, r *http.Request) {
var request struct{ Model string }
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
t.Error(err)
}
models <- request.Model
if calls.Add(1) == 1 {
close(entered)
select {
case <-release:
case <-r.Context().Done():
return
}
}
reply(w, validAnswer)
})
defer unblock()
snapshot := c.WithModel("test/preview-model")
activeCtx, activeCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer activeCancel()
activeDone := make(chan error, 1)
go func() {
_, err := c.Classify(activeCtx, f, d, true)
activeDone <- err
}()
select {
case <-entered:
case <-time.After(time.Second):
t.Fatal("first AI request did not start")
}
queueCtx, queueCancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
_, err := snapshot.Classify(queueCtx, f, d, true)
queueCancel()
if !errors.Is(err, context.DeadlineExceeded) || calls.Load() != 1 {
t.Fatalf("queued snapshot contacted provider or ignored cancellation: %v, attempts=%d", err, calls.Load())
}
unblock()
select {
case err := <-activeDone:
if err != nil {
t.Fatalf("queued cancellation disrupted active request: %v", err)
}
case <-time.After(time.Second):
t.Fatal("active request did not complete")
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
p, err := snapshot.Classify(ctx, f, d, true)
if err != nil || p.Enrichment.Classification.Model != "test/preview-model" || calls.Load() != 2 {
t.Fatalf("cancelled waiter leaked a request or blocked successor: %+v, %v, attempts=%d", p, err, calls.Load())
}
if original, preview := <-models, <-models; original != "test/strict-model" || preview != "test/preview-model" {
t.Fatalf("serialized requests used wrong models: %q, %q", original, preview)
}
}
func assertSafeRateLimit(t *testing.T, err error, provenance string) {
t.Helper()
if err == nil || !strings.Contains(err.Error(), "429") || provenance == "" || strings.Contains(err.Error(), "sensitive") || strings.Contains(provenance, "sensitive") {
t.Fatalf("unsafe or missing rate-limit failure: %v, provenance=%q", err, provenance)
}
}
type rateLimitRoundTripFunc func(*http.Request) (*http.Response, error)
func (f rateLimitRoundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) {
return f(r)
}
type rateLimitNotifyingBody struct {
io.ReadCloser
closed chan struct{}
once *sync.Once
}
func (b *rateLimitNotifyingBody) Close() error {
err := b.ReadCloser.Close()
b.once.Do(func() { close(b.closed) })
return err
}
+183
View File
@@ -0,0 +1,183 @@
// Package ratelimit coordinates bounded HTTP 429 retries for one provider client.
package ratelimit
import (
"context"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"sync"
"time"
)
const (
maxRateAttempts = 4
maxRateWait = 2 * time.Minute
)
// Controller serializes provider calls and retains their HTTP 429 cooldown.
// Its zero value is ready for use. A Controller must not be copied after use.
// The active caller owns retries; other callers fail fast during known cooldowns.
type Controller struct {
once sync.Once
active chan struct{}
mu sync.Mutex
limit *RateLimitError
}
// RateLimitError is a safe provider HTTP 429 error. Its message contains only
// the status and retry timing, never provider response text or request details.
// Use errors.As with *RateLimitError to identify it through wrapped errors.
type RateLimitError struct {
next time.Time
unbounded bool
}
func (r *RateLimitError) Error() string {
if r.unbounded {
return "provider rate limit (HTTP 429): retry time exceeds the supported range; automatic retry disabled"
}
return "provider rate limit (HTTP 429): retry allowed at " + r.next.UTC().Format(time.RFC3339Nano)
}
// RetryAt returns the earliest allowed retry time. Zero means the provider's
// delay exceeded the supported range and automatic retries remain disabled.
func (r *RateLimitError) RetryAt() time.Time {
return r.next
}
func (g *Controller) cooldown() error {
g.mu.Lock()
defer g.mu.Unlock()
if g.limit != nil && (g.limit.unbounded || time.Now().Before(g.limit.next)) {
return g.limit
}
return nil
}
// Acquire waits for the active call, unless canceled or a cooldown is known.
// A successful acquisition must be paired with Release, including on errors.
func (g *Controller) Acquire(ctx context.Context) error {
if err := ctx.Err(); err != nil {
return fmt.Errorf("provider request canceled: %w", err)
}
if err := g.cooldown(); err != nil {
return err
}
g.once.Do(func() { g.active = make(chan struct{}, 1) })
select {
case g.active <- struct{}{}:
case <-ctx.Done():
return fmt.Errorf("provider request canceled: %w", ctx.Err())
}
if err := ctx.Err(); err != nil {
g.Release()
return fmt.Errorf("provider request canceled: %w", err)
}
// The preceding request may have established a cooldown while we queued.
if err := g.cooldown(); err != nil {
g.Release()
return err
}
return nil
}
// Release relinquishes a successful acquisition.
func (g *Controller) Release() {
<-g.active
}
// retryLimit never converts a positive overflowing delay into a short wait.
// Delays beyond time.Duration's range disable retries rather than truncate the
// provider's instruction. HTTP dates retain their absolute timestamp unchanged.
func retryLimit(header string, now time.Time, fallback time.Duration) *RateLimitError {
limit := &RateLimitError{next: now.Add(fallback)}
header = strings.TrimSpace(header)
if header == "" {
return limit
}
digits := true
for _, c := range header {
if c < '0' || c > '9' {
digits = false
break
}
}
if digits {
seconds, err := strconv.ParseUint(header, 10, 64)
if err != nil || seconds > uint64((1<<63-1)/int64(time.Second)) {
return &RateLimitError{unbounded: true}
}
if delay := time.Duration(seconds) * time.Second; delay > fallback {
limit.next = now.Add(delay)
}
return limit
}
if date, err := http.ParseTime(header); err == nil && date.After(limit.next) {
limit.next = date
}
return limit
}
// Do executes an attempt under an already-acquired Controller. Only HTTP 429
// responses are retried, and only when retry is true (safe/idempotent requests).
// Each 429 body is closed here; other response bodies remain caller-owned.
// The callback must honor ctx and return errors safe to expose to the caller.
// Its per-attempt timeout must not include this controller's retry waiting.
func (g *Controller) Do(ctx context.Context, attempt func(context.Context) (*http.Response, error), retry bool) (*http.Response, error) {
remainingWait := maxRateWait
var lastLimit error
canceled := func(err error) error {
if lastLimit != nil {
return fmt.Errorf("%w: %w", lastLimit, err)
}
return fmt.Errorf("provider request canceled: %w", err)
}
for number := range maxRateAttempts {
if err := ctx.Err(); err != nil {
return nil, canceled(err)
}
resp, err := attempt(ctx)
if err != nil {
if ctx.Err() != nil {
return nil, canceled(ctx.Err())
}
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return nil, canceled(err)
}
return nil, err
}
if resp.StatusCode != http.StatusTooManyRequests {
return resp, nil
}
limit := retryLimit(resp.Header.Get("Retry-After"), time.Now(), time.Second<<number)
g.mu.Lock()
g.limit = limit
g.mu.Unlock()
// Never read or expose provider errors, and release each response before
// any sleep or retry. Other responses are processed by the caller.
resp.Body.Close()
lastLimit = limit
if err := ctx.Err(); err != nil {
return nil, canceled(err)
}
delay := time.Until(limit.next)
if !retry || number == maxRateAttempts-1 || limit.unbounded || delay > remainingWait {
return nil, limit
}
if delay <= 0 {
continue
}
remainingWait -= delay
timer := time.NewTimer(delay)
select {
case <-ctx.Done():
timer.Stop()
return nil, canceled(ctx.Err())
case <-timer.C:
}
}
return nil, lastLimit
}