diff --git a/OPERATIONS.txt b/OPERATIONS.txt index 24a7a4a..f91fcc3 100644 --- a/OPERATIONS.txt +++ b/OPERATIONS.txt @@ -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 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 -------------- Register your application and public certificate with Enable Banking. For diff --git a/README.md b/README.md index 9b580bf..985f4dc 100644 --- a/README.md +++ b/README.md @@ -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. +**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 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. +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 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. diff --git a/internal/app/app_test.go b/internal/app/app_test.go index 0de1afa..f8afecc 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -4,11 +4,14 @@ import ( "context" "encoding/hex" "encoding/json" + "errors" "net/http" "net/http/httptest" "reflect" "strings" + "sync/atomic" "testing" + "time" "finance-duck/internal/analytics" "finance-duck/internal/classification" @@ -86,6 +89,90 @@ func TestFailedClassificationStillImportsAndRetryIsIdempotent(t *testing.T) { 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)) { t.Helper() mock := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/app/backfill_test.go b/internal/app/backfill_test.go index 5dcfac3..c8085b2 100644 --- a/internal/app/backfill_test.go +++ b/internal/app/backfill_test.go @@ -22,10 +22,10 @@ type backfillBank struct { 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) if b.statusErr != nil { - return banking.Session{}, b.statusErr + return banking.SessionStatus{}, b.statusErr } return b.historyBank.Status(ctx, id) } @@ -204,7 +204,7 @@ func TestBackfillRejectsUnsafePrerequisitesBeforeProviderContact(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) { a, s, b := backfillApp(t) 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"}} case "partial retrieval": 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) if err != nil { @@ -233,7 +237,10 @@ func TestBackfillProviderFailuresDoNotImportPartialData(t *testing.T) { if strings.Contains(err.Error(), "private provider response") { 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") } current, err := a.Snapshot(context.Background()) diff --git a/internal/app/consent_test.go b/internal/app/consent_test.go index 64f37c0..84a6876 100644 --- a/internal/app/consent_test.go +++ b/internal/app/consent_test.go @@ -185,11 +185,11 @@ func TestRenewedConsentWakesSchedulerAndAutomaticallyImports(t *testing.T) { 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 { - 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) { diff --git a/internal/app/import.go b/internal/app/import.go index bd9a7cf..d0fdc79 100644 --- a/internal/app/import.go +++ b/internal/app/import.go @@ -12,6 +12,7 @@ import ( "finance-duck/internal/banking" "finance-duck/internal/classification" "finance-duck/internal/domain" + "finance-duck/internal/ratelimit" ) 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) if err != nil { - if errors.Is(err, banking.ErrReconnect) { - return ImportResult{}, banking.ErrReconnect - } - return ImportResult{}, errors.New("bank connection unavailable; retry importing history") + return ImportResult{}, bankFailure(err, "bank connection unavailable; retry importing history") } expiry, err = time.Parse(time.RFC3339, current.ValidUntil) if err != nil || !expiry.After(time.Now()) { return ImportResult{}, banking.ErrReconnect } - if !slices.ContainsFunc(current.Accounts, func(linked domain.Account) bool { - return linked.ExternalAccountID == account.ExternalAccountID - }) { + if !slices.Contains(current.AccountIDs, account.ExternalAccountID) { return ImportResult{}, banking.ErrReconnect } 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")) if err != nil { - if errors.Is(err, banking.ErrReconnect) { - return ImportResult{}, banking.ErrReconnect - } - return ImportResult{}, errors.New("transaction retrieval failed; retry importing history") + return ImportResult{}, bankFailure(err, "transaction retrieval failed; retry importing history") } // Use normal import processing without changing sync cursors or saved consent // 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") } + +// 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) { a.mu.Lock() defer a.mu.Unlock() @@ -343,25 +350,35 @@ func (a *App) Sync(ctx context.Context) (State, error) { } validAccounts := map[string]bool{} accountSession := map[string]string{} + failedSessions := map[string]bool{} for i, session := range a.ops.Sessions { for _, account := range session.Accounts { accountSession[account.ID] = session.ID } meta := a.ops.Consents[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 { - meta.Error = e.Error() + meta.Error = bankFailure(e, "bank connection unavailable; retry synchronization").Error() meta.NeedsReconnect = errors.Is(e, banking.ErrReconnect) a.ops.Consents[session.ID] = meta failures = append(failures, meta.Institution+": "+meta.Error) + failedSessions[session.ID] = true continue } meta.Error = "" meta.NeedsReconnect = false a.ops.Consents[session.ID] = meta a.ops.Sessions[i].ValidUntil = current.ValidUntil - for _, account := range current.Accounts { - validAccounts[account.ExternalAccountID] = true + for _, account := range session.Accounts { + if account.ExternalAccountID != "" && slices.Contains(current.AccountIDs, account.ExternalAccountID) { + validAccounts[account.ID] = true + } } } now := time.Now().UTC() @@ -370,8 +387,18 @@ func (a *App) Sync(ctx context.Context) (State, error) { if !account.Active || account.ExternalAccountID == "" { 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") + if sessionID != "" { + meta := a.ops.Consents[sessionID] + meta.Error = banking.ErrReconnect.Error() + meta.NeedsReconnect = true + a.ops.Consents[sessionID] = meta + } continue } 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) if e != nil { - meta := a.ops.Consents[accountSession[account.ID]] - meta.Error = "Transaction retrieval failed; retry synchronization" - a.ops.Consents[accountSession[account.ID]] = meta - failures = append(failures, account.DisplayName+": transaction retrieval failed") + meta := a.ops.Consents[sessionID] + meta.Error = bankFailure(e, "transaction retrieval failed; retry synchronization").Error() + meta.NeedsReconnect = meta.NeedsReconnect || errors.Is(e, banking.ErrReconnect) + a.ops.Consents[sessionID] = meta + failures = append(failures, account.DisplayName+": "+meta.Error) continue } result, e := a.importFacts(ctx, s, facts) diff --git a/internal/app/reclassify.go b/internal/app/reclassify.go index 34afffa..7a7d86a 100644 --- a/internal/app/reclassify.go +++ b/internal/app/reclassify.go @@ -70,7 +70,7 @@ func (a *App) Preview(ctx context.Context, r PreviewRequest) (Preview, error) { } a.mu.Lock() s, err := a.snapshot(ctx) - client := a.classifier + client := a.classifier.WithModel(r.Model) a.mu.Unlock() if err != nil { return Preview{}, err @@ -78,7 +78,6 @@ func (a *App) Preview(ctx context.Context, r PreviewRequest) (Preview, error) { if r.Revision != s.Revision { 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()} baseMerchants := len(s.Data.Merchants) for _, t := range s.Data.Transactions { @@ -90,6 +89,9 @@ func (a *App) Preview(ctx context.Context, r PreviewRequest) (Preview, error) { } p.Analysed++ proposal, e := client.Classify(ctx, t.Facts, s.Data, true) + if err = ctx.Err(); err != nil { + return Preview{}, err + } if e != nil { p.Errors = append(p.Errors, ClassificationError{t.Facts.ID, e.Error()}) continue diff --git a/internal/app/sync_test.go b/internal/app/sync_test.go index fa4edfb..c0f8321 100644 --- a/internal/app/sync_test.go +++ b/internal/app/sync_test.go @@ -3,12 +3,17 @@ package app import ( "context" "errors" + "fmt" + "io" + "net/http" "reflect" + "strings" "testing" "time" "finance-duck/internal/banking" "finance-duck/internal/domain" + "finance-duck/internal/ratelimit" ) 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) { 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 { - 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) { 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") } } + +// 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") + } + }) + } +} diff --git a/internal/banking/enablebanking.go b/internal/banking/enablebanking.go index 22ec103..8d90f67 100644 --- a/internal/banking/enablebanking.go +++ b/internal/banking/enablebanking.go @@ -23,6 +23,7 @@ import ( "time" "finance-duck/internal/domain" + "finance-duck/internal/ratelimit" ) // ErrReconnect identifies inactive bank consent, not application authentication @@ -34,6 +35,13 @@ type Session struct { ValidUntil string `json:"valid_until"` 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 { Amount domain.Money `json:"amount"` Currency string `json:"currency"` @@ -43,7 +51,7 @@ type Balance struct { type Provider interface { Authorize(context.Context, string, string, string) (string, 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) Transactions(context.Context, domain.Account, string, string) ([]domain.Facts, error) } @@ -53,6 +61,7 @@ type EnableBanking struct { appID string key *rsa.PrivateKey redirectURL string + requests ratelimit.Controller } var _ Provider = (*EnableBanking)(nil) @@ -125,9 +134,10 @@ func (p *EnableBanking) jwt() (string, error) { return unsigned + "." + base64.RawURLEncoding.EncodeToString(signature), nil } 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. - ctx, cancel := context.WithTimeout(ctx, 30*time.Second) - defer cancel() + if err := p.requests.Acquire(ctx); err != nil { + return fmt.Errorf("Enable Banking: %w", err) + } + defer p.requests.Release() token, err := p.jwt() if err != nil { return err @@ -157,17 +167,27 @@ func (p *EnableBanking) request(ctx context.Context, method, path string, input, if p.HTTPClient != nil { 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. 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 errors.Is(err, context.Canceled) { - return context.Canceled - } - if errors.Is(err, context.DeadlineExceeded) { - return fmt.Errorf("Enable Banking request timed out") - } - return fmt.Errorf("Enable Banking connection failed") + return fmt.Errorf("Enable Banking: %w", err) } defer response.Body.Close() 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 b, err := io.ReadAll(io.LimitReader(response.Body, limit+1)) 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") } if len(b) > limit { @@ -311,54 +337,34 @@ func (p *EnableBanking) Exchange(ctx context.Context, code string) (Session, err } 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 == "" { - return Session{}, fmt.Errorf("session ID is required") + return SessionStatus{}, fmt.Errorf("session ID is required") } var response struct { - Status string `json:"status"` - Accounts []string `json:"accounts"` - AccountsData []accountDTO `json:"accounts_data"` - Access accessDTO `json:"access"` - ASPSP institutionDTO `json:"aspsp"` + Status string `json:"status"` + Accounts []string `json:"accounts"` + Access accessDTO `json:"access"` } 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" { - 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) 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()) { - return Session{}, 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 + return SessionStatus{}, fmt.Errorf("Enable Banking session expired: %w", ErrReconnect) } for _, id := range response.Accounts { 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 { diff --git a/internal/banking/enablebanking_test.go b/internal/banking/enablebanking_test.go index 7dae5ca..400715a 100644 --- a/internal/banking/enablebanking_test.go +++ b/internal/banking/enablebanking_test.go @@ -13,11 +13,15 @@ import ( "encoding/pem" "errors" "fmt" + "io" "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "time" + + "finance-duck/internal/ratelimit" ) func testProvider(t *testing.T, handler http.HandlerFunc) (*EnableBanking, *rsa.PrivateKey) { @@ -138,7 +142,8 @@ func TestEnableBankingDocumentedFlowAndPagination(t *testing.T) { 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) 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": 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": @@ -180,8 +185,8 @@ func TestEnableBankingDocumentedFlowAndPagination(t *testing.T) { if err != nil { t.Fatal(err) } - if len(status.Accounts) != 1 || status.Accounts[0].ID != session.Accounts[0].ID || status.Accounts[0].ExternalAccountID != "uid-one" { - t.Fatalf("account identity changed between session DTOs: %+v", status) + if len(status.AccountIDs) != 1 || status.AccountIDs[0] != session.Accounts[0].ExternalAccountID || status.ValidUntil != expiry { + t.Fatalf("incorrect consent membership or expiry: %+v", status) } balances, err := p.Balances(context.Background(), "uid-one") 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) { _, key := testProvider(t, func(w http.ResponseWriter, r *http.Request) { t.Error("credential validation must not call provider") diff --git a/internal/classification/client.go b/internal/classification/client.go index 1f48f3c..ae45f45 100644 --- a/internal/classification/client.go +++ b/internal/classification/client.go @@ -11,18 +11,63 @@ import ( "net/http" "net/url" "strings" + "sync/atomic" "time" "unicode/utf8" "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 { APIKey string Model string IncludeAmount bool HTTPClient *http.Client 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 { @@ -41,9 +86,12 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D } } 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) { - p.Enrichment.Classification = domain.Provenance{Source: "fallback", Timestamp: time.Now().UTC().Format(time.RFC3339), Error: message} - return p, errors.New(message) + return failError(errors.New(message)) } if _, err := facts.Amount.Minor(); err != nil { 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 } } - 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") } + gate := c.rateControl() + if err := gate.Acquire(ctx); err != nil { + return failError(err) + } + defer gate.Release() clean := newSanitizer(facts, data, false) merchantClean := newSanitizer(facts, data, true) 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"` Currency string `json:"currency,omitempty"` }{Description: clean(facts.RawDescription), Categories: candidates.categories, Tags: candidates.tags, Merchants: candidates.merchants} - if c.IncludeAmount { + if includeAmount { prompt.Amount = &facts.Amount // 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 { @@ -91,7 +146,7 @@ func (c *Client) Classify(ctx context.Context, facts domain.Facts, data domain.D return fail("cannot encode classification request") } request := map[string]any{ - "model": c.Model, + "model": model, "stream": false, "max_tokens": 512, // 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 { return fail("cannot encode classification request") } - base := strings.TrimRight(c.BaseURL, "/") + base := strings.TrimRight(baseURL, "/") if base == "" { 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")) { 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} - if c.HTTPClient != nil { - client = *c.HTTPClient + if configuredHTTPClient != nil { + client = *configuredHTTPClient if client.Timeout == 0 { client.Timeout = 45 * time.Second } } // Redirects could send sensitive prompts to endpoints with different policies. 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 { - return fail("AI request failed") + return failError(err) } defer resp.Body.Close() 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 raw, err := io.ReadAll(io.LimitReader(resp.Body, maxResponse+1)) 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") } 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.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 if proposed != nil { validationData.Merchants = append(append([]domain.Merchant{}, data.Merchants...), *proposed) diff --git a/internal/classification/client_test.go b/internal/classification/client_test.go index 3244bb3..927b712 100644 --- a/internal/classification/client_test.go +++ b/internal/classification/client_test.go @@ -290,7 +290,7 @@ func TestUnsafeMerchantProposalRejected(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) { f, d := fixture() calls := 0 diff --git a/internal/classification/rate_limit_test.go b/internal/classification/rate_limit_test.go new file mode 100644 index 0000000..92575da --- /dev/null +++ b/internal/classification/rate_limit_test.go @@ -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 +} diff --git a/internal/ratelimit/controller.go b/internal/ratelimit/controller.go new file mode 100644 index 0000000..9e50c0d --- /dev/null +++ b/internal/ratelimit/controller.go @@ -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< 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 +}