nearle_daily/backend_fiesta
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

secert updated

Browse Source
This commit is contained in:
abhishek
2026-09-24 17:20:04 +05:30
parent 299871b820
commit 00317a00d8
11 changed files with 517 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

153
utils/embedding.go
View File

@@ -8,6 +8,8 @@ import (
"fmt"
"io"
"net/http"
"reflect"
"strconv"
"strings"
"time"
@@ -161,7 +163,41 @@ func (e *geminiEmbedder) Embed(ctx context.Context, text string) ([]float32, err
// so a rate-limited assistant told a shopkeeper "embedding: HTTP 429" — a
// sentence about a subsystem they have never heard of, describing something
// that was not involved.
// postJSON sends the request, and sends it a second time if the provider said
// it was over its quota and named a wait we are willing to hold for.
//
// One retry, not a loop: past that, a queue forms behind a limit that is not
// going to lift, and the person is better told to try again than left watching
// a spinner. Both the chat gateway and the embedder go through here, so neither
// can be the one that forgot.
func postJSON(ctx context.Context, client *http.Client, what, url, auth string, body, out interface{}, headers ...string) error {
err := postJSONOnce(ctx, client, what, url, auth, body, out, headers...)
var busy *tooManyRequests
if errors.As(err, &busy) && waitBeforeRetry(ctx, busy.after) {
// `out` must be emptied first. The first attempt decoded the provider's
// error body into it, and `encoding/json` leaves fields the second
// payload does not mention exactly as it found them — so a retry that
// SUCCEEDED came back carrying the 429's `error` object, and every
// caller here checks that field before the data. The result was a
// successful call reported as the failure it had just recovered from.
resetForRetry(out)
return postJSONOnce(ctx, client, what, url, auth, body, out, headers...)
}
return err
}
// resetForRetry empties a decode target so a second attempt cannot inherit the
// first one's fields.
func resetForRetry(out interface{}) {
value := reflect.ValueOf(out)
if value.Kind() != reflect.Ptr || value.IsNil() {
return
}
value.Elem().Set(reflect.Zero(value.Elem().Type()))
}
func postJSONOnce(ctx context.Context, client *http.Client, what, url, auth string, body, out interface{}, headers ...string) error {
payload, err := json.Marshal(body)
if err != nil {
return err
@@ -194,6 +230,20 @@ func postJSON(ctx context.Context, client *http.Client, what, url, auth string,
return fmt.Errorf("%s: HTTP %d, unreadable body: %w", what, resp.StatusCode, err)
}
if resp.StatusCode/100 != 2 {
// Too many requests is the one status that is not about this request.
// The provider's own sentence is unusable here — Groq's reads
//
// "Rate limit reached for model `openai/gpt-oss-120b` in organization
// `org_01m38x8s72e759kn6g88ve2dhj` service tier `on_demand` on tokens
// per minute (TPM): Limit 8000, Used 7320…"
//
// which is shown to a shopkeeper as Buddy's answer, names our billing
// account, and tells them nothing they can act on. It is also usually
// over within a second, so the honest handling is to wait and try again
// rather than to report it at all.
if resp.StatusCode == http.StatusTooManyRequests {
return &tooManyRequests{what: what, after: retryAfter(resp, raw)}
}
// The decoded body carries the provider's message where there is one;
// this is the fallback for a bare status.
if msg := extractMessage(raw); msg != "" {
@@ -230,3 +280,106 @@ func VectorLiteral(v []float32) string {
b.WriteByte(']')
return b.String()
}
// ── Being rate limited ──────────────────────────────────────────────────────
//
// A shared provider quota is not a fault in the request that happened to hit
// it, and on Groq's free tier it is reached by ordinary use: 8,000 tokens a
// minute is three or four Buddy questions. The waits are short — the provider
// states them in milliseconds — so one retry turns almost all of them into a
// slightly slower answer instead of an error.
// ErrBusy is what a caller sees when the wait did not help.
//
// Sentinel so the HTTP layer can answer 429 and the console can say "a moment"
// rather than rendering a provider's billing details as an answer.
var ErrBusy = errors.New("the assistant is busy right now — try again in a moment")
type tooManyRequests struct {
what string
after time.Duration
}
func (e *tooManyRequests) Error() string { return e.what + ": " + ErrBusy.Error() }
func (e *tooManyRequests) Unwrap() error { return ErrBusy }
// maxRetryWait bounds how long a request may be held. Beyond this the honest
// answer is "busy" — a person watching a spinner has already decided something
// is broken, and the provider's own suggestion can be a minute on a hard quota.
const maxRetryWait = 3 * time.Second
// retryAfter reads how long the provider asked us to wait.
//
// `Retry-After` first, because it is the standard and a proxy may add it where
// the body has nothing. Groq puts the number in prose instead — "Please try
// again in 840ms" — so that is read next. Zero means "no idea", and the caller
// uses its own floor rather than hammering immediately.
func retryAfter(resp *http.Response, raw []byte) time.Duration {
if header := strings.TrimSpace(resp.Header.Get("Retry-After")); header != "" {
// Seconds, as an integer, is the only form worth reading: the HTTP-date
// form is for caches and no model provider sends it.
if secs, err := strconv.ParseFloat(header, 64); err == nil && secs > 0 {
return time.Duration(secs * float64(time.Second))
}
}
return waitFromMessage(extractMessage(raw))
}
// waitFromMessage pulls "try again in 840ms" or "try again in 1.5s" out of prose.
//
// Its own function because it is the part worth testing: the wording comes from
// somebody else's error strings and is the first thing that will change.
func waitFromMessage(message string) time.Duration {
lower := strings.ToLower(message)
marker := "try again in "
at := strings.Index(lower, marker)
if at < 0 {
return 0
}
rest := lower[at+len(marker):]
end := 0
for end < len(rest) && (rest[end] == '.' || (rest[end] >= '0' && rest[end] <= '9')) {
end++
}
if end == 0 {
return 0
}
amount, err := strconv.ParseFloat(rest[:end], 64)
if err != nil || amount <= 0 {
return 0
}
switch {
case strings.HasPrefix(rest[end:], "ms"):
return time.Duration(amount * float64(time.Millisecond))
case strings.HasPrefix(rest[end:], "s"):
return time.Duration(amount * float64(time.Second))
}
return 0
}
// waitBeforeRetry sleeps for what the provider asked, bounded, and reports
// whether waiting is worth it at all.
//
// Returns false when the ask is longer than we are prepared to hold a request
// for, or when the caller's context is done — a retry after the browser has
// given up is work nobody will see.
func waitBeforeRetry(ctx context.Context, after time.Duration) bool {
if after <= 0 {
// No stated wait. A short one anyway: retrying instantly on a quota is
// how a burst becomes two bursts.
after = 250 * time.Millisecond
}
if after > maxRetryWait {
return false
}
timer := time.NewTimer(after)
defer timer.Stop()
select {
case <-ctx.Done():
return false
case <-timer.C:
return true
}
}

63
utils/embedding_test.go
View File

@@ -3,6 +3,7 @@ package utils
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"strings"
@@ -73,16 +74,70 @@ func TestGeminiEmbedderSendsTheRequestTheAPIExpects(t *testing.T) {
}
func TestEmbedderSurfacesProviderErrors(t *testing.T) {
// Used to assert this with a 429, which is now the one status that is
// deliberately NOT passed through — see the two tests below. The rule it was
// written for still holds for every other failure: a provider that explains
// itself should not have that explanation swallowed.
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(429)
w.Write([]byte(`{"error":{"message":"Rate limit reached"}}`))
w.WriteHeader(400)
w.Write([]byte(`{"error":{"message":"input exceeds the maximum token count"}}`))
}))
defer srv.Close()
e, _ := NewEmbedder(config.EmbeddingConfig{Provider: "openai", Model: "m", APIKey: "k", BaseURL: srv.URL})
_, err := e.Embed(context.Background(), "x")
if err == nil || !strings.Contains(err.Error(), "429") || !strings.Contains(err.Error(), "Rate limit") {
t.Fatalf("want a 429 with the provider's message, got %v", err)
if err == nil || !strings.Contains(err.Error(), "400") || !strings.Contains(err.Error(), "maximum token count") {
t.Fatalf("want a 400 with the provider's message, got %v", err)
}
}
func TestARateLimitIsWaitedOutRatherThanReported(t *testing.T) {
// The common case on a free tier: the quota clears in under a second, so
// the right answer is a slightly slower success rather than an error.
var calls int
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
calls++
if calls == 1 {
w.WriteHeader(429)
w.Write([]byte(`{"error":{"message":"Rate limit reached. Please try again in 20ms."}}`))
return
}
w.Write([]byte(`{"data":[{"embedding":[0.5,0.25]}]}`))
}))
defer srv.Close()
e, _ := NewEmbedder(config.EmbeddingConfig{Provider: "openai", Model: "m", APIKey: "k", BaseURL: srv.URL})
vector, err := e.Embed(context.Background(), "x")
if err != nil {
t.Fatalf("a 20ms quota wait became an error: %v", err)
}
if len(vector) != 2 {
t.Fatalf("the retry did not return the answer: %v", vector)
}
if calls != 2 {
t.Fatalf("expected one retry, saw %d calls", calls)
}
}
func TestAPersistentRateLimitIsReportedWithoutTheProvidersDetails(t *testing.T) {
// When waiting does not help, the person is told to try again — not handed
// our organisation id and the tokens-per-minute arithmetic behind it.
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(429)
w.Write([]byte(`{"error":{"message":"Rate limit reached for model X in organization org_01m38 on tokens per minute (TPM): Limit 8000. Please try again in 20ms."}}`))
}))
defer srv.Close()
e, _ := NewEmbedder(config.EmbeddingConfig{Provider: "openai", Model: "m", APIKey: "k", BaseURL: srv.URL})
_, err := e.Embed(context.Background(), "x")
if !errors.Is(err, ErrBusy) {
t.Fatalf("a persistent quota was not reported as busy: %v", err)
}
for _, leaked := range []string{"org_01m38", "TPM", "8000"} {
if strings.Contains(err.Error(), leaked) {
t.Fatalf("%q reached the caller: %s", leaked, err.Error())
}
}
}

131
utils/ratelimit_test.go Normal file
View File

@@ -0,0 +1,131 @@
package utils
import (
"context"
"errors"
"net/http"
"testing"
"time"
)
/*
Being rate limited by the model provider.
Groq's free tier is 8,000 tokens a minute, which is three or four Buddy
questions — so this is ordinary use, not an edge case. Its message reads:
Rate limit reached for model `openai/gpt-oss-120b` in organization
`org_01m38x8s72e759kn6g88ve2dhj` service tier `on_demand` on tokens per
minute (TPM): Limit 8000, Used 7320, Requested 792. Please try again in 840ms.
That sentence was being returned to the console as Buddy's ANSWER. It names our
billing account, it is about arithmetic the shopkeeper cannot influence, and the
condition it describes is usually over in under a second.
*/
func TestTheWaitIsReadOutOfTheProvidersProse(t *testing.T) {
// The part most likely to change, because it is somebody else's wording.
for _, tc := range []struct {
message string
want time.Duration
}{
{"Please try again in 840ms.", 840 * time.Millisecond},
{"please try again in 1.5s", 1500 * time.Millisecond},
{"Try again in 2s. Need more tokens?", 2 * time.Second},
{"Rate limit reached … Please try again in 397.499999ms.", 397499999 * time.Nanosecond},
} {
got := waitFromMessage(tc.message)
// Milliseconds is the resolution that matters; the fractional tail of
// 397.499999ms is noise.
if (got - tc.want).Abs() > time.Millisecond {
t.Fatalf("%q read as %v, want %v", tc.message, got, tc.want)
}
}
}
func TestProseWithNoWaitInItReadsAsUnknown(t *testing.T) {
// Zero means "no idea", and the caller uses its own floor. Guessing a
// number here would be inventing one.
for _, message := range []string{
"", "Rate limit reached.", "try again later", "try again in soon", "try again in 0ms",
} {
if got := waitFromMessage(message); got != 0 {
t.Fatalf("%q produced a wait of %v", message, got)
}
}
}
func TestTheRetryAfterHeaderWinsOverTheProse(t *testing.T) {
// It is the standard, and a proxy can add it where the body has nothing.
resp := &http.Response{Header: http.Header{}}
resp.Header.Set("Retry-After", "2")
got := retryAfter(resp, []byte(`{"error":{"message":"try again in 840ms"}}`))
if got != 2*time.Second {
t.Fatalf("header ignored: got %v", got)
}
}
func TestAnUnwaitableLimitIsNotWaitedFor(t *testing.T) {
// A hard quota can suggest a minute. Holding a request that long is worse
// than saying "busy" — the person watching the spinner decided it was
// broken long before it returned.
if waitBeforeRetry(context.Background(), time.Minute) {
t.Fatal("agreed to hold the request for a minute")
}
}
func TestNothingIsWaitedForOnceTheCallerHasGone(t *testing.T) {
// A retry after the browser gave up is work nobody will see.
ctx, cancel := context.WithCancel(context.Background())
cancel()
if waitBeforeRetry(ctx, 10*time.Millisecond) {
t.Fatal("waited although the request was already abandoned")
}
}
func TestAShortWaitIsHonoured(t *testing.T) {
start := time.Now()
if !waitBeforeRetry(context.Background(), 30*time.Millisecond) {
t.Fatal("refused a 30ms wait")
}
if time.Since(start) < 25*time.Millisecond {
t.Fatal("returned without waiting")
}
}
func TestBeingBusyIsRecognisableWithoutReadingTheText(t *testing.T) {
// The HTTP layer answers 429 on this, and the console tells the person to
// try again. Matching on the provider's wording instead would break the
// moment Groq rephrases it.
err := error(&tooManyRequests{what: "assistant", after: time.Second})
if !errors.Is(err, ErrBusy) {
t.Fatal("a rate-limited call is not recognisable as busy")
}
}
func TestTheProvidersBillingDetailsAreNotInTheMessage(t *testing.T) {
// The whole point. Whatever the provider said, this is what a shopkeeper
// reads.
err := error(&tooManyRequests{what: "assistant", after: 840 * time.Millisecond})
for _, leaked := range []string{"org_", "TPM", "8000", "tier", "billing"} {
if contains(err.Error(), leaked) {
t.Fatalf("%q reaches the console: %s", leaked, err.Error())
}
}
}
func contains(haystack, needle string) bool {
return len(needle) > 0 && len(haystack) >= len(needle) &&
func() bool {
for i := 0; i+len(needle) <= len(haystack); i++ {
if haystack[i:i+len(needle)] == needle {
return true
}
}
return false
}()
}