internal: various cleanup

Signed-off-by: NotAShelf <raf@notashelf.dev>
Change-Id: I48f7eac2e99d1d2af02fcf237401be096a6a6964
This commit is contained in:
raf 2026-05-09 13:24:46 +03:00
commit 56e1935ead
Signed by: NotAShelf
GPG key ID: 29D95B64378DB4BF
6 changed files with 33 additions and 48 deletions

23
internal/cache/db.go vendored
View file

@ -248,12 +248,12 @@ func (d *DB) SetNegative(storePath string, ttl time.Duration) error {
// Returns true if a non-expired negative entry exists for storePath. // Returns true if a non-expired negative entry exists for storePath.
func (d *DB) IsNegative(storePath string) (bool, error) { func (d *DB) IsNegative(storePath string) (bool, error) {
var count int var exists bool
err := d.db.QueryRow( err := d.db.QueryRow(
`SELECT COUNT(*) FROM negative_cache WHERE store_path = ? AND expires_at > ?`, `SELECT EXISTS(SELECT 1 FROM negative_cache WHERE store_path = ? AND expires_at > ?)`,
storePath, time.Now().Unix(), storePath, time.Now().Unix(),
).Scan(&count) ).Scan(&exists)
return count > 0, err return exists, err
} }
// Deletes expired negative cache entries. // Deletes expired negative cache entries.
@ -304,17 +304,10 @@ func (d *DB) LoadAllHealth() ([]HealthRow, error) {
// Deletes the oldest routes (by last_verified) when over capacity. // Deletes the oldest routes (by last_verified) when over capacity.
func (d *DB) evictIfNeeded() error { func (d *DB) evictIfNeeded() error {
count, err := d.RouteCount() _, err := d.db.Exec(`
if err != nil {
return err
}
if count <= d.maxEntries {
return nil
}
excess := count - d.maxEntries
_, err = d.db.Exec(`
DELETE FROM routes WHERE store_path IN ( DELETE FROM routes WHERE store_path IN (
SELECT store_path FROM routes ORDER BY last_verified ASC LIMIT ? SELECT store_path FROM routes ORDER BY last_verified ASC
)`, excess) LIMIT MAX(0, (SELECT COUNT(*) FROM routes) - ?)
)`, d.maxEntries)
return err return err
} }

View file

@ -140,9 +140,11 @@ func RunGossipLoop(node *Node, src RouteSource, peers []string, interval time.Du
continue continue
} }
for _, peer := range peers { for _, peer := range peers {
if err := Announce(peer, node, routes); err != nil { go func(p string) {
slog.Warn("mesh: announce failed", "peer", peer, "error", err) if err := Announce(p, node, routes); err != nil {
slog.Warn("mesh: announce failed", "peer", p, "error", err)
} }
}(peer)
} }
slog.Debug("mesh: announced routes to peers", "routes", len(routes), "peers", len(peers)) slog.Debug("mesh: announced routes to peers", "routes", len(routes), "peers", len(peers))
} }

View file

@ -150,10 +150,3 @@ func (rs *RouteStore) Top(n int) []cache.RouteEntry {
} }
return result return result
} }
func min(a, b int) int {
if a < b {
return a
}
return b
}

View file

@ -141,12 +141,7 @@ func (p *Prober) RecordFailure(url string) {
return return
} }
h.ConsecutiveFails++ h.ConsecutiveFails++
switch { h.Status = computeStatus(h.ConsecutiveFails)
case h.ConsecutiveFails >= 10:
h.Status = StatusDown
case h.ConsecutiveFails >= 3:
h.Status = StatusDegraded
}
if p.persistHealth != nil { if p.persistHealth != nil {
u, ema, cf, tq := h.URL, h.EMALatency, h.ConsecutiveFails, h.TotalQueries u, ema, cf, tq := h.URL, h.EMALatency, h.ConsecutiveFails, h.TotalQueries
fn := p.persistHealth fn := p.persistHealth
@ -199,9 +194,9 @@ func (p *Prober) ProbeUpstream(url string) {
// Skip if URL is not in table. This prevents in-flight probes from // Skip if URL is not in table. This prevents in-flight probes from
// resurrecting removed upstreams (race: RemoveUpstream called while // resurrecting removed upstreams (race: RemoveUpstream called while
// ProbeUpstream is in flight). // ProbeUpstream is in flight).
p.mu.Lock() p.mu.RLock()
_, exists := p.table[url] _, exists := p.table[url]
p.mu.Unlock() p.mu.RUnlock()
if !exists { if !exists {
// URL was removed or never added; do not resurrect. // URL was removed or never added; do not resurrect.
return return

View file

@ -152,7 +152,7 @@ func (r *Router) race(storeHash string, candidates []string) (*Result, error) {
return return
} }
resp.Body.Close() resp.Body.Close()
if resp.StatusCode != 200 { if resp.StatusCode != http.StatusOK {
mu.Lock() mu.Lock()
notFounds++ notFounds++
mu.Unlock() mu.Unlock()
@ -229,7 +229,7 @@ func (r *Router) fetchNarInfo(upstream, storeHash string) ([]byte, string, strin
return nil, "", "", 0 return nil, "", "", 0
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != 200 { if resp.StatusCode != http.StatusOK {
return nil, "", "", 0 return nil, "", "", 0
} }
body, err := io.ReadAll(resp.Body) body, err := io.ReadAll(resp.Body)

View file

@ -26,6 +26,7 @@ type Server struct {
upstreams []config.UpstreamConfig upstreams []config.UpstreamConfig
client *http.Client client *http.Client
cachePriority int cachePriority int
metricsHandler http.Handler
} }
// Creates a Server. // Creates a Server.
@ -37,6 +38,7 @@ func New(r *router.Router, p *prober.Prober, db *cache.DB, upstreams []config.Up
upstreams: upstreams, upstreams: upstreams,
client: &http.Client{Timeout: 60 * time.Second}, client: &http.Client{Timeout: 60 * time.Second},
cachePriority: cachePriority, cachePriority: cachePriority,
metricsHandler: promhttp.Handler(),
} }
} }
@ -48,7 +50,7 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
case path == "/health": case path == "/health":
s.handleHealth(w, r) s.handleHealth(w, r)
case path == "/metrics": case path == "/metrics":
promhttp.Handler().ServeHTTP(w, r) s.metricsHandler.ServeHTTP(w, r)
case strings.HasSuffix(path, ".narinfo"): case strings.HasSuffix(path, ".narinfo"):
s.handleNarinfo(w, r) s.handleNarinfo(w, r)
case strings.HasPrefix(path, "/nar/"): case strings.HasPrefix(path, "/nar/"):