kyle/sift
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

p3: tavily parse errors degrade; retry DNS on search

Browse Source 
A 200 with a bad Tavily body is tavily: error, not an empty success
that would cache for the hour. Upstream clients stay in the source
list if DNS is down at boot and retry the pin on Search; /healthz
lists them as unresolved until a pin lands.
This commit is contained in:
kyle
2026-09-23 12:22:31 -07:00
parent f748e695b1
commit 429f6b207e
8 changed files with 177 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/api.go
+5 -3
View File
@@ -110,9 +110,11 @@ func (s *Server) healthz(w http.ResponseWriter, _ *http.Request) {
if s.SelfIP != nil {
out["self_ip"] = s.SelfIP.Status()
}
if s.Search != nil && len(s.Search.Degraded) > 0 {
out["status"] = "degraded"
out["search"] = s.Search.Degraded
if s.Search != nil {
if d := s.Search.Health(); len(d) > 0 {
out["status"] = "degraded"
out["search"] = d
}
}
_ = json.NewEncoder(w).Encode(out)
}
internal/search/build.go
+26 -28
View File
@@ -17,7 +17,8 @@ const (
)
// Build constructs sources from config. Missing keys are listed on Engine.Degraded
// and logged once; those sources are not attempted.
// and logged once; those sources are not attempted. A source whose DNS is down
// at boot stays in the list and is retried on Search.
func Build(cfg config.Search, cache *dcache.DB, emb Embedder, log *slog.Logger) *Engine {
timeout := config.ParseDuration(cfg.Timeout, 8*time.Second)
ttl := config.ParseDuration(cfg.CacheTTL, time.Hour)
@@ -40,15 +41,9 @@ func Build(cfg config.Search, cache *dcache.DB, emb Embedder, log *slog.Logger)
if u == "" {
u = defaultSearx
}
c, err := dial.New(u, dial.Options{Loopback: true, Timeout: timeout})
if err != nil {
e.Degraded = append(e.Degraded, "search.searxng: "+err.Error())
if log != nil {
log.Error("search source", "source", NameSearxNG, "err", err)
}
} else {
e.Sources = append(e.Sources, &SearxNG{Client: c})
}
addPinned(e, log, NameSearxNG, u, dial.Options{Loopback: true, Timeout: timeout}, func(c *dial.Client) Source {
return &SearxNG{Client: c}
})
}
if cfg.BraveLLM.Enabled {
key := os.Getenv("BRAVE_API_KEY")
@@ -62,15 +57,9 @@ func Build(cfg config.Search, cache *dcache.DB, emb Embedder, log *slog.Logger)
if u == "" {
u = defaultBrave
}
c, err := dial.New(u, dial.Options{Timeout: timeout})
if err != nil {
e.Degraded = append(e.Degraded, "search.brave_llm: "+err.Error())
if log != nil {
log.Error("search source", "source", NameBraveLLM, "err", err)
}
} else {
e.Sources = append(e.Sources, &BraveLLM{Client: c, Key: key})
}
addPinned(e, log, NameBraveLLM, u, dial.Options{Timeout: timeout}, func(c *dial.Client) Source {
return &BraveLLM{Client: c, Key: key}
})
}
}
if cfg.Tavily.Enabled {
@@ -85,16 +74,25 @@ func Build(cfg config.Search, cache *dcache.DB, emb Embedder, log *slog.Logger)
if u == "" {
u = defaultTav
}
c, err := dial.New(u, dial.Options{Timeout: timeout})
if err != nil {
e.Degraded = append(e.Degraded, "search.tavily: "+err.Error())
if log != nil {
log.Error("search source", "source", NameTavily, "err", err)
}
} else {
e.Sources = append(e.Sources, &Tavily{Client: c, Key: key})
}
addPinned(e, log, NameTavily, u, dial.Options{Timeout: timeout}, func(c *dial.Client) Source {
return &Tavily{Client: c, Key: key}
})
}
}
return e
}
func addPinned(e *Engine, log *slog.Logger, name, raw string, opt dial.Options, wrap func(*dial.Client) Source) {
c, err := dial.New(raw, opt)
if err != nil {
e.Degraded = append(e.Degraded, "search."+name+": unavailable")
if log != nil {
log.Error("search source", "source", name, "err", "unavailable")
}
return
}
e.Sources = append(e.Sources, wrap(c))
if !c.Ready() && log != nil {
log.Error("search source unresolved, will retry", "source", name)
}
}
internal/search/dial/dial.go
+25 -6
View File
@@ -77,9 +77,7 @@ func New(raw string, opt Options) (*Client, error) {
lookup: lookup,
refresh: opt.Refresh,
}
if err := c.resolve(context.Background()); err != nil {
return nil, err
}
_ = c.resolve(context.Background()) // DNS down: keep the client, retry in Search
d := &net.Dialer{
Timeout: opt.Timeout,
Control: c.control,
@@ -204,17 +202,38 @@ func (c *Client) checkRedirect(req *http.Request, via []*http.Request) error {
return nil
}
// RefreshIfStale re-resolves when the pin is older than Refresh.
// RefreshIfStale re-resolves when unpinned or the pin is older than Refresh.
// A failed refresh leaves existing pins in place.
func (c *Client) RefreshIfStale(ctx context.Context) error {
c.mu.Lock()
stale := time.Since(c.resolved) >= c.refresh
need := len(c.ips) == 0 || time.Since(c.resolved) >= c.refresh
c.mu.Unlock()
if !stale {
if !need {
return nil
}
return c.resolve(ctx)
}
// Ensure resolves if needed. Error only when still unpinned afterwards.
func (c *Client) Ensure(ctx context.Context) error {
if c == nil {
return fmt.Errorf("search dial: nil")
}
if err := c.RefreshIfStale(ctx); err != nil && !c.Ready() {
return err
}
return nil
}
func (c *Client) Ready() bool {
if c == nil {
return false
}
c.mu.Lock()
defer c.mu.Unlock()
return len(c.ips) > 0
}
func (c *Client) IPs() []netip.Addr {
c.mu.Lock()
defer c.mu.Unlock()
internal/search/dial/dial_test.go
+52 -6
View File
@@ -2,16 +2,20 @@ package dial_test
import (
"context"
"errors"
"net/http"
"net/http/httptest"
"net/netip"
"net/url"
"strings"
"sync/atomic"
"testing"
"git.wntrmute.dev/kyle/sift/internal/search/dial"
)
var errDNS = errors.New("dns down")
func TestLoopbackAllowsConfiguredPort(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(200)
@@ -53,25 +57,31 @@ func TestLoopbackRefusesOtherPort(t *testing.T) {
}
func TestPublicPoisonedPrivateRefused(t *testing.T) {
_, err := dial.New("https://api.tavily.com/search", dial.Options{
c, err := dial.New("https://api.tavily.com/search", dial.Options{
Loopback: false,
Lookup: func(_ context.Context, host string) ([]netip.Addr, error) {
return []netip.Addr{netip.MustParseAddr("127.0.0.1")}, nil
},
})
if err == nil {
t.Fatal("poisoned A record allowed")
if err != nil {
t.Fatal(err)
}
if c.Ready() {
t.Fatal("poisoned A record pinned")
}
}
func TestPublicPrivateRFC1918Refused(t *testing.T) {
_, err := dial.New("https://api.search.brave.com/res/v1/llm/context", dial.Options{
c, err := dial.New("https://api.search.brave.com/res/v1/llm/context", dial.Options{
Lookup: func(_ context.Context, _ string) ([]netip.Addr, error) {
return []netip.Addr{netip.MustParseAddr("10.0.0.5")}, nil
},
})
if err == nil {
t.Fatal("private A allowed")
if err != nil {
t.Fatal(err)
}
if c.Ready() {
t.Fatal("private A pinned")
}
}
@@ -110,3 +120,39 @@ func TestParseRejectsNonHTTP(t *testing.T) {
}
_ = url.URL{}
}
func TestResolveRetriesAfterDNSDown(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(200)
}))
defer ts.Close()
u, _ := url.Parse(ts.URL)
ip := netip.MustParseAddr(u.Hostname())
var n atomic.Int32
c, err := dial.New(ts.URL, dial.Options{
Loopback: true,
Lookup: func(_ context.Context, host string) ([]netip.Addr, error) {
if n.Add(1) == 1 {
return nil, errDNS
}
return []netip.Addr{ip}, nil
},
})
if err != nil {
t.Fatal(err)
}
if c.Ready() {
t.Fatal("first lookup should fail")
}
if err := c.Ensure(context.Background()); err != nil {
t.Fatal(err)
}
if !c.Ready() {
t.Fatal("retry")
}
resp, err := c.HTTP.Get(ts.URL + "/")
if err != nil {
t.Fatal(err)
}
resp.Body.Close()
}
internal/search/engine.go
+22 -1
View File
@@ -20,7 +20,7 @@ type Engine struct {
CacheTTL time.Duration
Embedder Embedder
Log *slog.Logger
Degraded []string // healthz: "search.brave_llm: missing key"
Degraded []string // static: missing keys
TavilyCap int
Now func() time.Time
@@ -46,6 +46,27 @@ type RunResult struct {
Attempted []string
}
type readySrc interface {
Name() string
Ready() bool
}
// Health is missing keys plus sources that still have no pin.
func (e *Engine) Health() []string {
if e == nil {
return nil
}
out := append([]string(nil), e.Degraded...)
for _, s := range e.Sources {
r, ok := s.(readySrc)
if !ok || r.Ready() {
continue
}
out = append(out, "search."+r.Name()+": unresolved")
}
return out
}
func (e *Engine) now() time.Time {
if e != nil && e.Now != nil {
return e.Now()
internal/search/sources.go
+18 -2
View File
@@ -30,6 +30,16 @@ func warnErr(name string, err error) string {
return name + ": error"
}
func pinErr(ctx context.Context, c *dial.Client, name string) string {
if c == nil {
return name + ": error"
}
if err := c.Ensure(ctx); err != nil {
return warnErr(name, err)
}
return ""
}
func readJSON(resp *http.Response, dest any) error {
defer resp.Body.Close()
if resp.StatusCode != 200 {
@@ -46,12 +56,16 @@ type SearxNG struct {
}
func (s *SearxNG) Name() string { return NameSearxNG }
func (s *SearxNG) Ready() bool { return s != nil && s.Client.Ready() }
func (s *BraveLLM) Ready() bool { return s != nil && s.Client.Ready() }
func (s *SearxNG) Search(ctx context.Context, query string, k int) SourceOut {
if s == nil || s.Client == nil {
return SourceOut{Warn: NameSearxNG + ": error"}
}
_ = s.Client.RefreshIfStale(ctx)
if w := pinErr(ctx, s.Client, NameSearxNG); w != "" {
return SourceOut{Warn: w}
}
base := strings.TrimRight(s.Client.URL.String(), "/")
u, err := url.Parse(base + "/search")
if err != nil {
@@ -110,7 +124,9 @@ func (s *BraveLLM) Search(ctx context.Context, query string, k int) SourceOut {
if s == nil || s.Client == nil {
return SourceOut{Warn: NameBraveLLM + ": error"}
}
_ = s.Client.RefreshIfStale(ctx)
if w := pinErr(ctx, s.Client, NameBraveLLM); w != "" {
return SourceOut{Warn: w}
}
u, err := url.Parse(s.Client.URL.String())
if err != nil {
return SourceOut{Warn: warnErr(NameBraveLLM, err)}
internal/search/sources_test.go
+18
View File
@@ -95,3 +95,21 @@ func TestBraveMapsSources(t *testing.T) {
t.Fatal("grounding used as snippet")
}
}
func TestTavilyBadJSONIsError(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write([]byte(`not-json`))
}))
defer ts.Close()
c, err := dial.New(ts.URL, dial.Options{Loopback: true, Timeout: time.Second})
if err != nil {
t.Fatal(err)
}
out := (&Tavily{Client: c, Key: "k"}).Search(context.Background(), "q", 5)
if out.Warn != "tavily: error" {
t.Fatalf("warn %q", out.Warn)
}
if len(out.Hits) != 0 || out.Credits != 0 {
t.Fatalf("%+v", out)
}
}
internal/search/tavily.go
+11 -5
View File
@@ -17,6 +17,7 @@ type Tavily struct {
}
func (s *Tavily) Name() string { return NameTavily }
func (s *Tavily) Ready() bool { return s != nil && s.Client.Ready() }
type tavilyReq struct {
Query string `json:"query"`
@@ -31,7 +32,9 @@ func (s *Tavily) Search(ctx context.Context, query string, k int) SourceOut {
if s == nil || s.Client == nil {
return SourceOut{Warn: NameTavily + ": error"}
}
_ = s.Client.RefreshIfStale(ctx)
if w := pinErr(ctx, s.Client, NameTavily); w != "" {
return SourceOut{Warn: w}
}
if k <= 0 {
k = DefaultK
}
@@ -66,11 +69,14 @@ func (s *Tavily) Search(ctx context.Context, query string, k int) SourceOut {
if err != nil {
return SourceOut{Warn: warnErr(NameTavily, err)}
}
hits, credits := mapTavily(raw)
hits, credits, err := mapTavily(raw)
if err != nil {
return SourceOut{Warn: NameTavily + ": error"}
}
return SourceOut{Hits: hits, Credits: credits}
}
func mapTavily(raw []byte) ([]Hit, int) {
func mapTavily(raw []byte) ([]Hit, int, error) {
var env struct {
Answer string `json:"answer"`
Results []struct {
@@ -84,7 +90,7 @@ func mapTavily(raw []byte) ([]Hit, int) {
} `json:"usage"`
}
if err := json.Unmarshal(raw, &env); err != nil {
return nil, 0
return nil, 0, err
}
_ = env.Answer // dropped; never returned
hits := make([]Hit, 0, len(env.Results))
@@ -98,5 +104,5 @@ func mapTavily(raw []byte) ([]Hit, int) {
if credits <= 0 {
credits = 1
}
return hits, credits
return hits, credits, nil
}